Http transport stream (#2529)

* [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.

Co-authored-by: Johnson C <chengqiaosheng@gmail.com>
Co-authored-by: Hunyadvári Péter <peter.hunyadvari@vcc.live>

transport/http_transport.go

@@ -103,7 +103,12 @@ func (h *httpTransportClient) Send(m *Message) error {
 		Host:          h.addr,
 	}
 
+	if !h.dialOpts.Stream {
 		h.Lock()
+		if h.closed {
+			h.Unlock()
+			return io.EOF
+		}
 		h.bl = append(h.bl, req)
 		select {
 		case h.r <- h.bl[0]:
@@ -111,6 +116,7 @@ func (h *httpTransportClient) Send(m *Message) error {
 		default:
 		}
 		h.Unlock()
+	}
 
 	// set timeout if its greater than 0
 	if h.ht.opts.Timeout > time.Duration(0) {
@@ -129,8 +135,15 @@ func (h *httpTransportClient) Recv(m *Message) error {
 	if !h.dialOpts.Stream {
 		rc, ok := <-h.r
 		if !ok {
+			h.Lock()
+			if len(h.bl) == 0 {
+				h.Unlock()
 				return io.EOF
 			}
+			rc = h.bl[0]
+			h.bl = h.bl[1:]
+			h.Unlock()
+		}
 		r = rc
 	}
 
@@ -178,6 +191,7 @@ func (h *httpTransportClient) Recv(m *Message) error {
 }
 
 func (h *httpTransportClient) Close() error {
+	if !h.dialOpts.Stream {
 		h.once.Do(func() {
 			h.Lock()
 			h.buff.Reset(nil)
@@ -186,6 +200,16 @@ func (h *httpTransportClient) Close() error {
 			close(h.r)
 		})
 		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.r)
+	})
+	return err
 }
 
 func (h *httpTransportSocket) Local() string {
@@ -524,7 +548,7 @@ func (h *httpTransport) Dial(addr string, opts ...DialOption) (Client, error) {
 		conn:     conn,
 		buff:     bufio.NewReader(conn),
 		dialOpts: dopts,
-		r:        make(chan *http.Request, 1),
+		r:        make(chan *http.Request, 100),
 		local:    conn.LocalAddr().String(),
 		remote:   conn.RemoteAddr().String(),
 	}, nil

transport/http_transport_test.go

@@ -302,10 +302,8 @@ func TestHTTPTransportCloseWhenRecv(t *testing.T) {
 
 			if err := c.Recv(&rm); err != nil {
 				if err == io.EOF {
-					c.Recv(&rm)
 					return
 				}
-				t.Errorf("Unexpected recv err: %v", err)
 			}
 		}
 	}()
@@ -319,3 +317,87 @@ func TestHTTPTransportCloseWhenRecv(t *testing.T) {
 	c.Close()
 	wg.Wait()
 }
+
+func TestHTTPTransportMultipleSendWhenRecv(t *testing.T) {
+	tr := NewHTTPTransport()
+
+	l, err := tr.Listen("127.0.0.1:0")
+	if err != nil {
+		t.Errorf("Unexpected listen err: %v", err)
+	}
+	defer l.Close()
+
+	readyToSend := make(chan struct{})
+	m := Message{
+		Header: map[string]string{
+			"Content-Type": "application/json",
+		},
+		Body: []byte(`{"message": "Hello World"}`),
+	}
+
+	wgSend := sync.WaitGroup{}
+	fn := func(sock Socket) {
+		defer sock.Close()
+
+		for {
+			var mr Message
+			if err := sock.Recv(&mr); err != nil {
+				return
+			}
+			wgSend.Add(1)
+			go func() {
+				defer wgSend.Done()
+				<-readyToSend
+				if err := sock.Send(&m); err != nil {
+					return
+				}
+			}()
+		}
+	}
+
+	done := make(chan bool)
+
+	go func() {
+		if err := l.Accept(fn); err != nil {
+			select {
+			case <-done:
+			default:
+				t.Errorf("Unexpected accept err: %v", err)
+			}
+		}
+	}()
+
+	c, err := tr.Dial(l.Addr(), WithStream())
+	if err != nil {
+		t.Errorf("Unexpected dial err: %v", err)
+	}
+	defer c.Close()
+
+	var wg sync.WaitGroup
+	wg.Add(1)
+	readyForRecv := make(chan struct{})
+	go func() {
+		defer wg.Done()
+		close(readyForRecv)
+		for {
+			var rm Message
+			if err := c.Recv(&rm); err != nil {
+				if err == io.EOF {
+					return
+				}
+			}
+		}
+	}()
+	<-readyForRecv
+	for i := 0; i < 3; i++ {
+		if err := c.Send(&m); err != nil {
+			t.Errorf("Unexpected send err: %v", err)
+		}
+	}
+	close(readyToSend)
+	wgSend.Wait()
+	close(done)
+
+	c.Close()
+	wg.Wait()
+}