Get rid of the worker goroutine

Author: viciious

connection.go

@@ -116,8 +116,6 @@ func connect(ctx context.Context, scheme, addr string, opts Options) (conn *Conn
 	// remove deadline
 	conn.tcpConn.SetDeadline(time.Time{})
 
-	go conn.worker()
-
 	return
 }
 
@@ -490,65 +488,29 @@ func (conn *Connection) setError(err error) {
 	}
 }
 
-func (conn *Connection) worker() {
-	var wg sync.WaitGroup
-
-	wg.Add(2)
-
-	go func() {
-		err := conn.writer()
-		conn.setError(err)
-		conn.stop()
-		wg.Done()
-	}()
-
-	go func() {
-		err := conn.reader()
-		conn.setError(err)
-		conn.stop()
-		wg.Done()
-	}()
-
-	wg.Wait()
-
-	// release all pending packets
-	writeChan := conn.writeChan
-
-CLEANUP_LOOP:
-	for {
-		select {
-		case req := <-writeChan:
-			pp := req.packet
-			if pp != nil {
-				req.packet = nil
-				conn.releasePacket(pp)
-			}
-		default:
-			break CLEANUP_LOOP
-		}
-	}
-
-	// send error reply to all pending requests
-	conn.requests.CleanUp(func(req *request) {
-		select {
-		case req.replyChan <- &AsyncResult{
-			Error:     ConnectionClosedError(conn),
-			ErrorCode: ErrNoConnection,
-			Opaque:    req.opaque,
-		}:
-		default:
-		}
-		requestPool.Put(req)
-	})
-
-	close(conn.closed)
-}
-
 func (conn *Connection) writer() (err error) {
 	writeChan := conn.writeChan
 	stopChan := conn.exit
 	w := bufio.NewWriterSize(conn.ccw, DefaultWriterBufSize)
 
+	defer close(conn.closed)
+
+	defer func() {
+	CLEANUP_LOOP:
+		for {
+			select {
+			case req := <-writeChan:
+				pp := req.packet
+				if pp != nil {
+					req.packet = nil
+					conn.releasePacket(pp)
+				}
+			default:
+				break CLEANUP_LOOP
+			}
+		}
+	}()
+
 	wr := func(w io.Writer, req *request) error {
 		packet := req.packet
 
@@ -604,6 +566,25 @@ func (conn *Connection) reader() (err error) {
 	var pp *BinaryPacket
 	var requestID uint64
 
+	defer func() {
+		<-conn.closed
+	}()
+
+	defer func() {
+		// send error reply to all pending requests
+		conn.requests.CleanUp(func(req *request) {
+			select {
+			case req.replyChan <- &AsyncResult{
+				Error:     ConnectionClosedError(conn),
+				ErrorCode: ErrNoConnection,
+				Opaque:    req.opaque,
+			}:
+			default:
+			}
+			requestPool.Put(req)
+		})
+	}()
+
 	r := bufio.NewReaderSize(conn.ccr, DefaultReaderBufSize)
 
 READER_LOOP: