| 27 | } |
| 28 | |
| 29 | func (s *serdes[_, R, RR]) recvLoop() { |
| 30 | s.logger.Debug(s.ctx, "starting recvLoop") |
| 31 | defer s.closeIdempotent() |
| 32 | defer close(s.recvCh) |
| 33 | for { |
| 34 | var length uint32 |
| 35 | if err := binary.Read(s.conn, binary.BigEndian, &length); err != nil { |
| 36 | s.logger.Debug(s.ctx, "failed to read length", slog.Error(err)) |
| 37 | return |
| 38 | } |
| 39 | if length > MaxLength { |
| 40 | s.logger.Critical(s.ctx, "message length exceeds max", |
| 41 | slog.F("length", length)) |
| 42 | return |
| 43 | } |
| 44 | s.logger.Debug(s.ctx, "about to read message", slog.F("length", length)) |
| 45 | mb := make([]byte, length) |
| 46 | if n, err := io.ReadFull(s.conn, mb); err != nil { |
| 47 | s.logger.Debug(s.ctx, "failed to read message", |
| 48 | slog.Error(err), |
| 49 | slog.F("num_bytes_read", n)) |
| 50 | return |
| 51 | } |
| 52 | msg := R(new(RR)) |
| 53 | if err := proto.Unmarshal(mb, msg); err != nil { |
| 54 | s.logger.Critical(s.ctx, "failed to unmarshal message", slog.Error(err)) |
| 55 | return |
| 56 | } |
| 57 | select { |
| 58 | case s.recvCh <- msg: |
| 59 | s.logger.Debug(s.ctx, "passed received message to speaker") |
| 60 | case <-s.ctx.Done(): |
| 61 | s.logger.Debug(s.ctx, "recvLoop canceled", slog.Error(s.ctx.Err())) |
| 62 | } |
| 63 | } |
| 64 | } |
| 65 | |
| 66 | func (s *serdes[S, _, _]) sendLoop() { |
| 67 | s.logger.Debug(s.ctx, "starting sendLoop") |