Add ConnectionController

integration_testing/connection_bridge.go

@@ -142,13 +142,12 @@ func (bridge *ConnectionBridge) createOutboundConnection(node *cmd.Node, otherNo
 
 		na, err := lib.IPToNetAddr(conn.RemoteAddr().String(), otherNode.Server.GetConnectionManager().AddrMgr,
 			otherNode.Params)
-		messagesFromPeer := make(chan *lib.ServerMessage)
+		messagesFromPeer := make(chan *lib.ServerMessage, 100)
 		newPeerChan := make(chan *lib.Peer, 100)
 		donePeerChan := make(chan *lib.Peer, 100)
 		peer := lib.NewPeer(uint64(lib.RandInt64(math.MaxInt64)), uint64(lib.RandInt64(math.MaxInt64)), conn,
 			false, na, false, 10000, 0, bridge.nodeB.Params,
 			messagesFromPeer, nil, nil, lib.NodeSyncTypeAny, newPeerChan, donePeerChan)
-		peer.ID = uint64(lib.RandInt64(math.MaxInt64))
 		bridge.newPeerChan <- peer
 		//}
 	}(ll)
@@ -183,10 +182,90 @@ func (bridge *ConnectionBridge) getVersionMessage(node *cmd.Node) *lib.MsgDeSoVe
 	return ver
 }
 
+func ReadWithTimeout(readFunc func() error, readTimeout time.Duration) error {
+	errChan := make(chan error)
+	go func() {
+		errChan <- readFunc()
+	}()
+	select {
+	case err := <-errChan:
+		{
+			return err
+		}
+	case <-time.After(readTimeout):
+		{
+			return fmt.Errorf("ReadWithTimeout: Timed out reading message")
+		}
+	}
+}
+
 // startConnection starts the connection by performing version and verack exchange with
 // the provided connection, pretending to be the otherNode.
 func (bridge *ConnectionBridge) startConnection(connection *lib.Peer, otherNode *cmd.Node) error {
-	// TODO: Update this
+	// Prepare the version message.
+	versionMessage := bridge.getVersionMessage(otherNode)
+	connection.VersionNonceSent = versionMessage.Nonce
+
+	// Send the version message.
+	fmt.Println("Sending version message:", versionMessage, versionMessage.StartBlockHeight)
+	if err := connection.WriteDeSoMessage(versionMessage); err != nil {
+		return err
+	}
+
+	// Wait for a response to the version message.
+	if err := ReadWithTimeout(
+		func() error {
+			msg, err := connection.ReadDeSoMessage()
+			if err != nil {
+				return err
+			}
+
+			verMsg, ok := msg.(*lib.MsgDeSoVersion)
+			if !ok {
+				return err
+			}
+
+			connection.VersionNonceReceived = verMsg.Nonce
+			connection.TimeConnected = time.Unix(verMsg.TstampSecs, 0)
+			connection.TimeOffsetSecs = verMsg.TstampSecs - time.Now().Unix()
+			return nil
+		}, lib.DeSoMainnetParams.VersionNegotiationTimeout); err != nil {
+
+		return err
+	}
+
+	// Now prepare the verack message.
+	verackMsg := lib.NewMessage(lib.MsgTypeVerack)
+	verackMsg.(*lib.MsgDeSoVerack).NonceReceived = connection.VersionNonceReceived
+
+	// And send it to the connection.
+	if err := connection.WriteDeSoMessage(verackMsg); err != nil {
+		return err
+	}
+
+	// And finally wait for connection's response to the verack message.
+	if err := ReadWithTimeout(
+		func() error {
+			msg, err := connection.ReadDeSoMessage()
+			if err != nil {
+				return err
+			}
+
+			if msg.GetMsgType() != lib.MsgTypeVerack {
+				return fmt.Errorf("message is not verack! Type: %v", msg.GetMsgType())
+			}
+			verackMsg := msg.(*lib.MsgDeSoVerack)
+			if verackMsg.NonceReceived != connection.VersionNonceSent {
+				return fmt.Errorf("verack message nonce doesn't match (received: %v, sent: %v)",
+					verackMsg.NonceReceived, connection.VersionNonceSent)
+			}
+			return nil
+		}, lib.DeSoMainnetParams.VersionNegotiationTimeout); err != nil {
+
+		return err
+	}
+	connection.VersionNegotiated = true
+
 	return nil
 }