fix rpcforward build err

rpcforward.go

@@ -67,37 +67,43 @@ func (cs *clientStream) SendMsgForward(mBytes []byte) (err error) {
 		return err
 	}
 	err = cs.withRetry(op, func() { cs.bufferForRetryLocked(len(hdr)+len(payload), op) })
-	if cs.binlog != nil && err == nil {
-		cs.binlog.Log(&binarylog.ClientMessage{
+	if len(cs.binlogs) != 0 && err == nil {
+		cm := &binarylog.ClientMessage{
 			OnClientSide: true,
 			Message:      msgBytes,
-		})
+		}
+		for _, binlog := range cs.binlogs {
+			binlog.Log(cm)
+		}
 	}
-	return
+	return err
 }
 func (cs *clientStream) RecvMsgForward() (mBytes []byte, err error) {
-	if cs.binlog != nil && !cs.serverHeaderBinlogged {
+	if len(cs.binlogs) != 0 && !cs.serverHeaderBinlogged {
 		// Call Header() to binary log header if it's not already logged.
 		cs.Header()
 	}
 	var recvInfo *payloadInfo
-	if cs.binlog != nil {
+	if len(cs.binlogs) != 0 {
 		recvInfo = &payloadInfo{}
 	}
 	mBytes, err = cs.withRetryForward(func(a *csAttempt) ([]byte, error) {
 		return a.recvMsgForward(recvInfo)
 	}, cs.commitAttemptLocked)
-	if cs.binlog != nil && err == nil {
-		cs.binlog.Log(&binarylog.ServerMessage{
+	if len(cs.binlogs) != 0 && err == nil {
+		sm := &binarylog.ServerMessage{
 			OnClientSide: true,
 			Message:      recvInfo.uncompressedBytes,
-		})
+		}
+		for _, binlog := range cs.binlogs {
+			binlog.Log(sm)
+		}
 	}
 	if err != nil || !cs.desc.ServerStreams {
 		// err != nil or non-server-streaming indicates end of stream.
 		cs.finish(err)
 
-		if cs.binlog != nil {
+		if len(cs.binlogs) != 0 {
 			// finish will not log Trailer. Log Trailer here.
 			logEntry := &binarylog.ServerTrailer{
 				OnClientSide: true,
@@ -107,10 +113,12 @@ func (cs *clientStream) RecvMsgForward() (mBytes []byte, err error) {
 			if logEntry.Err == io.EOF {
 				logEntry.Err = nil
 			}
-			if fromContext, ok := peer.FromContext(cs.Context()); ok {
-				logEntry.PeerAddr = fromContext.Addr
+			if peer, ok := peer.FromContext(cs.Context()); ok {
+				logEntry.PeerAddr = peer.Addr
+			}
+			for _, binlog := range cs.binlogs {
+				binlog.Log(logEntry)
 			}
-			cs.binlog.Log(logEntry)
 		}
 	}
 	return mBytes, err
@@ -420,17 +428,23 @@ func (ss *serverStream) SendMsgForward(req []byte) (err error) {
 	if err := ss.t.Write(ss.s, hdr, payload, &transport.Options{Last: false}); err != nil {
 		return toRPCErr(err)
 	}
-	if ss.binlog != nil {
+	if len(ss.binlogs) != 0 {
 		if !ss.serverHeaderBinlogged {
 			h, _ := ss.s.Header()
-			ss.binlog.Log(&binarylog.ServerHeader{
+			sh := &binarylog.ServerHeader{
 				Header: h,
-			})
+			}
 			ss.serverHeaderBinlogged = true
+			for _, binlog := range ss.binlogs {
+				binlog.Log(sh)
+			}
 		}
-		ss.binlog.Log(&binarylog.ServerMessage{
+		sm := &binarylog.ServerMessage{
 			Message: data,
-		})
+		}
+		for _, binlog := range ss.binlogs {
+			binlog.Log(sm)
+		}
 	}
 	if len(ss.statsHandler) != 0 {
 		for _, sh := range ss.statsHandler {
@@ -469,14 +483,17 @@ func (ss *serverStream) RecvMsgForward() (msgBytes []byte, err error) {
 		}
 	}()
 	var payInfo *payloadInfo
-	if len(ss.statsHandler) != 0 || ss.binlog != nil {
+	if len(ss.statsHandler) != 0 || len(ss.binlogs) != 0 {
 		payInfo = &payloadInfo{}
 	}
 	msgBytes, err = recvForward(ss.p, ss.codec, ss.s, ss.dc, ss.maxReceiveMessageSize, payInfo, ss.decomp)
 	if err != nil {
 		if err == io.EOF {
-			if ss.binlog != nil {
-				ss.binlog.Log(&binarylog.ClientHalfClose{})
+			if len(ss.binlogs) != 0 {
+				chc := &binarylog.ClientHalfClose{}
+				for _, binlog := range ss.binlogs {
+					binlog.Log(chc)
+				}
 			}
 			return nil, err
 		}
@@ -497,10 +514,13 @@ func (ss *serverStream) RecvMsgForward() (msgBytes []byte, err error) {
 			})
 		}
 	}
-	if ss.binlog != nil {
-		ss.binlog.Log(&binarylog.ClientMessage{
+	if len(ss.binlogs) != 0 {
+		cm := &binarylog.ClientMessage{
 			Message: payInfo.uncompressedBytes,
-		})
+		}
+		for _, binlog := range ss.binlogs {
+			binlog.Log(cm)
+		}
 	}
 	return msgBytes, nil
 }