pkg/p2p(cdc): implement MessageClient (#3668)

errors.toml

@@ -621,6 +621,16 @@ error = '''
 etcd api call error
 '''
 
+["CDC:ErrPeerMessageClientClosed"]
+error = '''
+peer-to-peer message client has been closed
+'''
+
+["CDC:ErrPeerMessageClientPermanentFail"]
+error = '''
+peer-to-peer message client has failed permanently, no need to reconnect: %s
+'''
+
 ["CDC:ErrPeerMessageDataLost"]
 error = '''
 peer-to-peer message data lost, topic: %s, seq: %s
@@ -636,6 +646,11 @@ error = '''
 peer-to-peer message duplicate connection: epoch %d
 '''
 
+["CDC:ErrPeerMessageEncodeError"]
+error = '''
+failed to encode peer-to-peer message
+'''
+
 ["CDC:ErrPeerMessageIllegalClientVersion"]
 error = '''
 peer-to-peer message client reported illegal version: %s
@@ -656,6 +671,11 @@ error = '''
 peer-to-peer message receiver is a mismatch: expected %s, got %s
 '''
 
+["CDC:ErrPeerMessageSendTryAgain"]
+error = '''
+peer-to-peer message client has too many pending messages to send, try again later
+'''
+
 ["CDC:ErrPeerMessageServerClosed"]
 error = '''
 peer-to-peer message server has closed connection: %s.

pkg/errors/errors.go

@@ -265,6 +265,10 @@ var (
 
 	// p2p error
 	ErrPeerMessageIllegalMeta          = errors.Normalize("peer-to-peer message server received an RPC call with illegal metadata", errors.RFCCodeText("CDC:ErrPeerMessageIllegalMeta"))
+	ErrPeerMessageClientPermanentFail  = errors.Normalize("peer-to-peer message client has failed permanently, no need to reconnect: %s", errors.RFCCodeText("CDC:ErrPeerMessageClientPermanentFail"))
+	ErrPeerMessageClientClosed         = errors.Normalize("peer-to-peer message client has been closed", errors.RFCCodeText("CDC:ErrPeerMessageClientClosed"))
+	ErrPeerMessageSendTryAgain         = errors.Normalize("peer-to-peer message client has too many pending messages to send, try again later", errors.RFCCodeText("CDC:ErrPeerMessageSendTryAgain"))
+	ErrPeerMessageEncodeError          = errors.Normalize("failed to encode peer-to-peer message", errors.RFCCodeText("CDC:ErrPeerMessageEncodeError"))
 	ErrPeerMessageInternalSenderClosed = errors.Normalize("peer-to-peer message server tries to send to a closed stream. Internal only.", errors.RFCCodeText("CDC:ErrPeerMessageInternalSenderClosed"))
 	ErrPeerMessageStaleConnection      = errors.Normalize("peer-to-peer message stale connection: old-epoch %d, new-epoch %d", errors.RFCCodeText("CDC:ErrPeerMessageStaleConnection"))
 	ErrPeerMessageDuplicateConnection  = errors.Normalize("peer-to-peer message duplicate connection: epoch %d", errors.RFCCodeText("CDC:ErrPeerMessageDuplicateConnection"))