Merge pull request #37 from IceFireDB/fix/performance-regression

Fix/performance regression

Author: gitsrc

redhub.go

@@ -102,13 +102,12 @@ func (rs *RedHub) OnClose(c gnet.Conn, err error) (action gnet.Action) {
 
 // OnTraffic fires when a socket receives data from the remote
 func (rs *RedHub) OnTraffic(c gnet.Conn) (action gnet.Action) {
-	var out []byte
 	rs.connSync.RLock()
 	cb, ok := rs.redHubBufMap[c]
 	rs.connSync.RUnlock()
 
 	if !ok {
-		c.AsyncWrite(resp.AppendError(nil, "ERR Client is closed"), nil)
+		_, _ = c.Write(resp.AppendError(nil, "ERR Client is closed"))
 		return gnet.None
 	}
 
@@ -120,28 +119,32 @@ func (rs *RedHub) OnTraffic(c gnet.Conn) (action gnet.Action) {
 	cb.buf.Write(buf)
 	cmds, lastbyte, err := resp.ReadCommands(cb.buf.Bytes())
 	if err != nil {
-		c.AsyncWrite(resp.AppendError(nil, "ERR "+err.Error()), nil)
+		_, _ = c.Write(resp.AppendError(nil, "ERR "+err.Error()))
 		return gnet.None
 	}
 
 	cb.command = append(cb.command, cmds...)
 	cb.buf.Reset()
 
 	if len(lastbyte) == 0 {
+		var out []byte
 		for len(cb.command) > 0 {
 			cmd := cb.command[0]
 			cb.command = cb.command[1:]
 
 			var status Action
-			result, status := rs.handler(cmd, out)
-			if len(result) > 0 {
-				c.AsyncWrite(result, nil)
-			}
+			out, status = rs.handler(cmd, out)
 
 			if status == Close {
+				if len(out) > 0 {
+					_, _ = c.Write(out)
+				}
 				return gnet.Close
 			}
 		}
+		if len(out) > 0 {
+			_, _ = c.Write(out)
+		}
 	} else {
 		cb.buf.Write(lastbyte)
 	}

redhub_test.go

@@ -24,6 +24,14 @@ func (m *mockConn) Write(buf []byte) (n int, err error) {
 	return len(buf), nil
 }
 
+func (m *mockConn) Writev(bufs [][]byte) (n int, err error) {
+	for _, buf := range bufs {
+		m.written = append(m.written, buf...)
+		n += len(buf)
+	}
+	return n, nil
+}
+
 func (m *mockConn) Close() error {
 	m.closed = true
 	return nil