Handle msg before sending to neighbors to prevent duplicate tx

Signed-off-by: Yilun <[email protected]>
nknorg · Nov 6, 2018 · 59049e2 · 59049e2
1 parent 2c51dc0
commit 59049e2
Show file tree

Hide file tree

Showing 4 changed files with 88 additions and 106 deletions.
diff --git a/net/message/message.go b/net/message/message.go
@@ -73,17 +73,11 @@ func AllocMsg(t string, length int) Messenger {
 	case "tx":
 		var msg trn
 		copy(msg.msgHdr.CMD[0:len(t)], t)
-		//if (message.Payload.Length <= 1024 * 1024)
-		//OnInventoryReceived(Transaction.DeserializeFrom(message.Payload));
 		return &msg
 	case "ising":
 		var msg IsingMessage
 		copy(msg.msgHdr.CMD[0:len(t)], t)
 		return &msg
-	case "txnpool":
-		var msg txnPool
-		copy(msg.msgHdr.CMD[0:len(t)], t)
-		return &msg
 	case "notfound":
 		var msg notFound
 		copy(msg.msgHdr.CMD[0:len(t)], t)

diff --git a/net/message/transaction.go b/net/message/transaction.go
@@ -145,20 +145,3 @@ func (msg *trn) Deserialize(r io.Reader) error {
 
 	return nil
 }
-
-type txnPool struct {
-	msgHdr
-	//TBD
-}
-
-func ReqTxnPool(node Noder) error {
-	msg := AllocMsg("txnpool", 0)
-	buff := bytes.NewBuffer(nil)
-	err := msg.Serialize(buff)
-	if err != nil {
-		return err
-	}
-	node.Tx(buff.Bytes())
-
-	return nil
-}
diff --git a/net/node/hashcache.go b/net/node/hashcache.go
@@ -7,8 +7,8 @@ import (
 )
 
 const (
-	HashCacheExpiration      = 60 * time.Second
-	HashCacheCleanupInterval = 1 * time.Second
+	HashCacheExpiration      = 300 * time.Second
+	HashCacheCleanupInterval = 10 * time.Second
 )
 
 type hashCache struct {

diff --git a/net/node/node.go b/net/node/node.go
@@ -175,109 +175,114 @@ func InitNode(pubKey *crypto.PubKey, nn *nnet.NNet) (Noder, error) {
 		return true
 	}))
 
-	nn.MustApplyMiddleware(nnetnode.BytesReceived(func(msg, msgID, srcID []byte, remoteNode *nnetnode.RemoteNode) ([]byte, bool) {
-		nbr := n
-		if remoteNode != nil {
-			nbr = n.getNbrByNNetNode(remoteNode)
-			if nbr == nil {
-				err := n.AddRemoteNode(remoteNode)
-				if err != nil {
-					log.Error("Cannot add remote node:", err)
-					return msg, true
-				}
+	nn.MustApplyMiddleware(routing.RemoteMessageRouted(func(remoteMessage *nnetnode.RemoteMessage, localNode *nnetnode.LocalNode, remoteNodes []*nnetnode.RemoteNode) (*nnetnode.RemoteMessage, *nnetnode.LocalNode, []*nnetnode.RemoteNode, bool) {
+		if remoteMessage.Msg.MessageType == nnetprotobuf.BYTES {
+			msgBody := &nnetprotobuf.Bytes{}
+			err := proto.Unmarshal(remoteMessage.Msg.Message, msgBody)
+			if err != nil {
+				log.Error(err)
+				return nil, nil, nil, false
+			}
 
-				nbr = n.getNbrByNNetNode(remoteNode)
-				if nbr == nil {
-					log.Error("Cannot get neighbor node")
-					return msg, true
+			if localNode != nil {
+				nbr := n
+				if remoteMessage.RemoteNode != nil {
+					nbr = n.getNbrByNNetNode(remoteMessage.RemoteNode)
+					if nbr == nil {
+						err = n.AddRemoteNode(remoteMessage.RemoteNode)
+						if err != nil {
+							log.Error("Cannot add remote node:", err)
+							return nil, nil, nil, false
+						}
+
+						nbr = n.getNbrByNNetNode(remoteMessage.RemoteNode)
+						if nbr == nil {
+							log.Error("Cannot get neighbor node")
+							return nil, nil, nil, false
+						}
+					}
 				}
-			}
-		}
 
-		err := message.HandleNodeMsg(nbr, msg)
-		if err != nil {
-			log.Error(err)
-			return msg, true
-		}
+				err = message.HandleNodeMsg(nbr, msgBody.Data)
+				if err != nil {
+					log.Error(err)
+					return nil, nil, nil, false
+				}
 
-		return msg, true
-	}))
+				if len(remoteNodes) == 0 {
+					return nil, nil, nil, false
+				}
 
-	nn.MustApplyMiddleware(routing.RemoteMessageRouted(func(remoteMessage *nnetnode.RemoteMessage, localNode *nnetnode.LocalNode, remoteNodes []*nnetnode.RemoteNode) (*nnetnode.RemoteMessage, *nnetnode.LocalNode, []*nnetnode.RemoteNode, bool) {
-		if remoteMessage.Msg.MessageType == nnetprotobuf.BYTES && remoteMessage.Msg.RoutingType == nnetprotobuf.RELAY {
-			if localNode != nil {
-				return remoteMessage, localNode, remoteNodes, false
+				localNode = nil
 			}
 
-			if len(remoteNodes) == 0 {
-				log.Error("No next hop found")
-				return nil, nil, nil, false
-			}
+			if remoteMessage.Msg.RoutingType == nnetprotobuf.RELAY {
+				if len(remoteNodes) > 1 {
+					log.Error("Multiple next hop is not supported yet")
+					return nil, nil, nil, false
+				}
 
-			if len(remoteNodes) > 1 {
-				log.Error("Multiple next hop is not supported yet")
-				return nil, nil, nil, false
-			}
+				nextHop := n.getNbrByNNetNode(remoteNodes[0])
+				if nextHop == nil {
+					err := n.AddRemoteNode(remoteNodes[0])
+					if err != nil {
+						log.Error("Cannot add next hop remote node:", err)
+						return nil, nil, nil, false
+					}
+
+					nextHop = n.getNbrByNNetNode(remoteNodes[0])
+					if nextHop == nil {
+						log.Error("Cannot get next hop neighbor node")
+						return nil, nil, nil, false
+					}
+				}
 
-			nextHop := n.getNbrByNNetNode(remoteNodes[0])
-			if nextHop == nil {
-				err := n.AddRemoteNode(remoteNodes[0])
+				msgBody := &nnetprotobuf.Bytes{}
+				err := proto.Unmarshal(remoteMessage.Msg.Message, msgBody)
 				if err != nil {
-					log.Error("Cannot add next hop remote node:", err)
+					log.Error(err)
 					return nil, nil, nil, false
 				}
 
-				nextHop = n.getNbrByNNetNode(remoteNodes[0])
-				if nextHop == nil {
-					log.Error("Cannot get next hop neighbor node")
+				msg, err := message.ParseMsg(msgBody.Data)
+				if err != nil {
+					log.Error(err)
 					return nil, nil, nil, false
 				}
-			}
 
-			msgBody := &nnetprotobuf.Bytes{}
-			err := proto.Unmarshal(remoteMessage.Msg.Message, msgBody)
-			if err != nil {
-				log.Error(err)
-				return nil, nil, nil, false
-			}
-
-			msg, err := message.ParseMsg(msgBody.Data)
-			if err != nil {
-				log.Error(err)
-				return nil, nil, nil, false
-			}
-
-			relayMsg, ok := msg.(*message.RelayMessage)
-			if !ok {
-				log.Error("Msg is not relay message")
-				return nil, nil, nil, false
-			}
+				relayMsg, ok := msg.(*message.RelayMessage)
+				if !ok {
+					log.Error("Msg is not relay message")
+					return nil, nil, nil, false
+				}
 
-			relayPacket := &relayMsg.Packet
-			err = n.relayer.SignRelayPacket(nextHop, relayPacket)
-			if err != nil {
-				log.Error(err)
-				return nil, nil, nil, false
-			}
+				relayPacket := &relayMsg.Packet
+				err = n.relayer.SignRelayPacket(nextHop, relayPacket)
+				if err != nil {
+					log.Error(err)
+					return nil, nil, nil, false
+				}
 
-			relayMsg, err = message.NewRelayMessage(relayPacket)
-			if err != nil {
-				log.Error(err)
-				return nil, nil, nil, false
-			}
+				relayMsg, err = message.NewRelayMessage(relayPacket)
+				if err != nil {
+					log.Error(err)
+					return nil, nil, nil, false
+				}
 
-			msgBody.Data, err = relayMsg.ToBytes()
-			if err != nil {
-				log.Error(err)
-				return nil, nil, nil, false
-			}
+				msgBody.Data, err = relayMsg.ToBytes()
+				if err != nil {
+					log.Error(err)
+					return nil, nil, nil, false
+				}
 
-			remoteMessage.Msg.Message, err = proto.Marshal(msgBody)
-			if err != nil {
-				log.Error(err)
-				return nil, nil, nil, false
+				remoteMessage.Msg.Message, err = proto.Marshal(msgBody)
+				if err != nil {
+					log.Error(err)
+					return nil, nil, nil, false
+				}
 			}
 		}
+
 		return remoteMessage, localNode, remoteNodes, true
 	}))