prevent closing before publishing ack (#42)

2026-09-30 23:47:22 +00:00 · 2023-03-10 01:57:45 -08:00
parent d134fc2a90
commit 6de16dbb52
1 changed files with 3 additions and 0 deletions
@@ -120,6 +120,9 @@ func (s *streamImpl[SendType, RecvType]) handleStream(is *internal.Stream) error
 			return ErrStreamClosed
 		}

+		s.pending.Inc()
+		defer s.pending.Dec()
+
 		v, err := deserializePayload[RecvType](b.Message.RawMessage, b.Message.Message)
 		if err != nil {
 			err = NewError(MalformedRequest, err)