Fix stream eos error (#2716)

* [fix] etcd config source prefix issue (#2389) * http transport data race issue (#2436) * [fix] #2431 http transport data race issue * [feature] Ability to close connection while receiving. Ability to send messages while receiving. Icreased r channel limit to 100 to more fluently communication. Do not dropp sent request if r channel is full. * [fix] Do not send error when stream send eos header, just close the connection * [fix] Do not overwrite the error in http client --------- Co-authored-by: Johnson C <[email protected]>
micro · Jul 2, 2024 · 1515db5 · 1515db5
1 parent 4d5b3b0
commit 1515db5
Show file tree

Hide file tree

Showing 3 changed files with 41 additions and 22 deletions.
diff --git a/server/rpc_server.go b/server/rpc_server.go
@@ -205,7 +205,9 @@ func (s *rpcServer) ServeConn(sock transport.Socket) {
 		// If we don't have a socket and its a stream
 		// Check if its a last stream EOS error
 		if !ok && stream && msg.Header[headers.Error] == errLastStreamResponse.Error() {
+			closeConn = true
 			pool.Release(psock)
+
 			continue
 		}
 
@@ -512,7 +514,11 @@ func (s *rpcServer) Deregister() error {
 			logger.Logf(log.InfoLevel, "Unsubscribing %s from topic: %s", node.Id, sub.Topic())
 
 			if err := sub.Unsubscribe(); err != nil {
-				logger.Logf(log.ErrorLevel, "Failed to unsubscribe subscriber nr. %d from topic %s: %v", i+1, sub.Topic(), err)
+				logger.Logf(log.ErrorLevel,
+					"Failed to unsubscribe subscriber nr. %d from topic %s: %v",
+					i+1,
+					sub.Topic(),
+					err)
 			}
 		}
 
@@ -759,7 +765,11 @@ Loop:
 
 			rerr := s.opts.RegisterCheck(s.opts.Context)
 			if rerr != nil && registered {
-				logger.Logf(log.ErrorLevel, "Server %s-%s register check error: %s, deregister it", config.Name, config.Id, rerr)
+				logger.Logf(log.ErrorLevel,
+					"Server %s-%s register check error: %s, deregister it",
+					config.Name,
+					config.Id,
+					rerr)
 				// deregister self in case of error
 				if err := s.Deregister(); err != nil {
 					logger.Logf(log.ErrorLevel, "Server %s-%s deregister error: %s", config.Name, config.Id, err)
@@ -809,7 +819,11 @@ Loop:
 	s.setOptsAddr(addr)
 }
 
-func (s *rpcServer) serveReq(ctx context.Context, msg transport.Message, req *rpcRequest, resp *rpcResponse, rcodec codec.Codec) {
+func (s *rpcServer) serveReq(ctx context.Context,
+	msg transport.Message,
+	req *rpcRequest,
+	resp *rpcResponse,
+	rcodec codec.Codec) {
 	logger := s.opts.Logger
 	router := s.getRouter()
 

diff --git a/transport/http_client.go b/transport/http_client.go
@@ -147,8 +147,8 @@ func (h *httpTransportClient) Recv(msg *Message) (err error) {
 	}
 
 	defer func() {
-		if err = rsp.Body.Close(); err != nil {
-			err = errors.Wrap(err, "failed to close body")
+		if err2 := rsp.Body.Close(); err2 != nil {
+			err = errors.Wrap(err2, "failed to close body")
 		}
 	}()
 
@@ -180,25 +180,29 @@ func (h *httpTransportClient) Recv(msg *Message) (err error) {
 
 func (h *httpTransportClient) Close() error {
 	if !h.dialOpts.Stream {
-		h.once.Do(func() {
-			h.Lock()
-			h.buff.Reset(nil)
-			h.closed = true
-			h.Unlock()
-			close(h.req)
-		})
+		h.once.Do(
+			func() {
+				h.Lock()
+				h.buff.Reset(nil)
+				h.closed = true
+				h.Unlock()
+				close(h.req)
+			},
+		)
 
 		return h.conn.Close()
 	}
 
 	err := h.conn.Close()
-	h.once.Do(func() {
-		h.Lock()
-		h.buff.Reset(nil)
-		h.closed = true
-		h.Unlock()
-		close(h.req)
-	})
+	h.once.Do(
+		func() {
+			h.Lock()
+			h.buff.Reset(nil)
+			h.closed = true
+			h.Unlock()
+			close(h.req)
+		},
+	)
 
 	return err
 }
diff --git a/transport/http_transport_test.go b/transport/http_transport_test.go
@@ -1,6 +1,7 @@
 package transport
 
 import (
+	"errors"
 	"io"
 	"net"
 	"sync"
@@ -126,7 +127,7 @@ func TestHTTPTransportError(t *testing.T) {
 		for {
 			var m Message
 			if err := sock.Recv(&m); err != nil {
-				if err == io.EOF {
+				if errors.Is(err, io.EOF) {
 					return
 				}
 				t.Fatal(err)
@@ -335,7 +336,7 @@ func TestHTTPTransportMultipleSendWhenRecv(t *testing.T) {
 		Body: []byte(`{"message": "Hello World"}`),
 	}
 
-	wgSend := sync.WaitGroup{}
+	var wgSend sync.WaitGroup
 	fn := func(sock Socket) {
 		defer sock.Close()
 
@@ -344,7 +345,6 @@ func TestHTTPTransportMultipleSendWhenRecv(t *testing.T) {
 			if err := sock.Recv(&mr); err != nil {
 				return
 			}
-			wgSend.Add(1)
 			go func() {
 				defer wgSend.Done()
 				<-readyToSend
@@ -388,6 +388,7 @@ func TestHTTPTransportMultipleSendWhenRecv(t *testing.T) {
 			}
 		}
 	}()
+	wgSend.Add(3)
 	<-readyForRecv
 	for i := 0; i < 3; i++ {
 		if err := c.Send(&m); err != nil {