Every node is now allowed to store some extra data that all of the nodes will have access to (The data is immutable for now, maybe later a broadcast message of new data will be made).

Author: dondish

cluster_test.go

@@ -20,7 +20,7 @@ func (t TestMessage) Type() string {
 }
 
 func TestCreateSingleNodeCluster(t *testing.T) {
-	master := CreateCluster("localhost:5555")
+	master := CreateCluster("localhost:5555", nil)
 	defer master.Close()
 
 	assert.Empty(t, master.Nodes, "the node should not be connected to other nodes implicitly")
@@ -29,19 +29,20 @@ func TestCreateSingleNodeCluster(t *testing.T) {
 }
 
 func TestCreateTwoNodeCluster(t *testing.T) {
-	master := CreateCluster("localhost:5556")
+	master := CreateCluster("localhost:5556", nil)
 	defer master.Close()
 	time.Sleep(500 * time.Millisecond)
-	node, err := JoinCluster("localhost:5557", "localhost:5556")
+	node, err := JoinCluster("localhost:5557", "localhost:5556", nil)
 
 	if err != nil {
 		fmt.Println("couldn't create node:", err)
 		assert.NotNil(t, err, "There shouldn't be an error while closing the slave")
 	}
 
 	defer node.Close()
+	time.Sleep(500 * time.Millisecond)
 	_, ok := node.Nodes.Load(0)
-	assert.True(t, ok, 0, "the node should have master in its nodes map")
+	assert.True(t, ok, "the node should have master in its nodes map")
 	assert.True(t, node.Id == 1, "the node's id should be set to 1")
 	assert.NotNil(t, master.Message, "the message channel should not be nil")
 
@@ -72,10 +73,10 @@ func TestCreateTwoNodeCluster(t *testing.T) {
 }
 
 func TestCreateMultiNodeCluster(t *testing.T) {
-	master := CreateCluster("localhost:5558")
+	master := CreateCluster("localhost:5558", nil)
 	defer master.Close()
 	time.Sleep(500 * time.Millisecond)
-	node1, err := JoinCluster("localhost:5559", "localhost:5558")
+	node1, err := JoinCluster("localhost:5559", "localhost:5558", nil)
 
 	if err != nil {
 		fmt.Println("couldn't create node 1:", err)
@@ -84,7 +85,7 @@ func TestCreateMultiNodeCluster(t *testing.T) {
 
 	defer node1.Close()
 	time.Sleep(500 * time.Millisecond)
-	node2, err := JoinCluster("localhost:5560", "localhost:5558")
+	node2, err := JoinCluster("localhost:5560", "localhost:5558", nil)
 
 	if err != nil {
 		fmt.Println("couldn't create node 2:", err)

connection.go

@@ -1,3 +1,4 @@
+// TODO Add heartbeat mechanism
 package go_cluster
 
 import (
@@ -8,11 +9,12 @@ import (
 )
 
 // Resembles a connection between two nodes.
-// The API makes it simpler to customize the connection.
 type Connection struct {
-	Conn *net.TCPConn
+	Conn *net.TCPConn // The connection
+	Data interface{}  // The remote node's extra data
 }
 
+// The transport protocol
 type Transport struct {
 	Type    string
 	Message Message
@@ -32,15 +34,17 @@ func connect(address string) (*Connection, error) {
 }
 
 // Connects a node to a new node and sets up message handling
-func connectNewNode(id int, addr string, node *Node) {
+func connectNewNode(id int, addr string, data interface{}, node *Node) {
 	if conn, err := connect(addr); err != nil {
 		node.Log("error while connecting to a new node:", err)
 	} else {
+		conn.Data = data
 		node.Nodes.Store(id, conn)
-		msg := IdReqMessage{Id: node.Id}
+		msg := GreetingMessage{Id: node.Id, Data: data}
 		if err := conn.Write(msg); err != nil {
 			node.Log("couldn't send the message to the new node:", err)
 		}
+		go handleMessages(conn, node, id)
 		node.Log("Successfully connected to the new node!")
 	}
 }
@@ -70,18 +74,19 @@ func handleIncoming(address string, node *Node) {
 			continue
 		}
 		connection := &Connection{Conn: conn}
+		node.Nodes.Store(node.NextId, connection)
 		go handleMessages(connection, node, node.NextId)
-		if err := connection.Write(ReadyMessage{Id: node.NextId}); err != nil {
+		if err := connection.Write(ReadyMessage{Id: node.NextId, EntryId: node.Id}); err != nil {
 			node.Log("failed to send ready message to ", conn.RemoteAddr().String(), ":", err.Error())
 			continue
 		}
-		node.Nodes.Store(node.NextId, connection)
 		node.NextId++
 	}
 }
 
 // Handles the messages incoming from the connection
 func handleMessages(connection *Connection, node *Node, remoteid int) {
+	node.Log("Message handling started with", remoteid, "(note that this might not be the ")
 	conn := connection.Conn
 	dec := gob.NewDecoder(conn)
 	for {
@@ -104,22 +109,36 @@ func handleMessages(connection *Connection, node *Node, remoteid int) {
 				node.Message <- message
 			}
 		} else {
-			if data.Type == "idreq" {
-				remoteid = data.Message.(IdReqMessage).Id
+			if data.Type == "readyreq" {
+				if _, ok := node.Nodes.Load(remoteid); !ok {
+					readymsg := data.Message.(ReadyMessage)
+					node.Id = readymsg.Id
+					remoteid = readymsg.EntryId
+					node.NextId = node.Id + 1
+					if err := connection.Write(IntroduceMessage{Addr: node.Addr, Data: data}); err != nil {
+						node.Log("couldn't send an introduce message to the entry point:", err)
+						os.Exit(1)
+					}
+					node.Nodes.Store(remoteid, connection)
+					node.Log("Node Intialized!")
+				}
+			} else if data.Type == "greetreq" {
+				remoteid = data.Message.(GreetingMessage).Id
 				node.Log("Successfully connected to node", remoteid)
 				node.Nodes.Store(remoteid, connection)
-			} else if data.Type == "addrreq" {
+			} else if data.Type == "introreq" {
 				msg := NewNodeMessage{
 					Id:   remoteid,
-					Addr: data.Message.(AddressMessage).Addr,
+					Addr: data.Message.(IntroduceMessage).Addr,
+					Data: data.Message.(IntroduceMessage).Data,
 				}
 				if err := node.Broadcast(msg, remoteid); err != nil {
 					node.Log("error while broadcasting a new node:", err)
 				}
-			} else if data.Type == "newnode" {
+			} else if data.Type == "newnodereq" {
 				msg := data.Message.(NewNodeMessage)
 				node.NextId = msg.Id + 1
-				connectNewNode(msg.Id, msg.Addr, node)
+				connectNewNode(msg.Id, msg.Addr, msg.Data, node)
 			} else {
 				node.Message <- data.Message
 			}

message.go

@@ -2,7 +2,6 @@ package go_cluster
 
 import (
 	"encoding/gob"
-	"strconv"
 )
 
 // The Message interface, this is supposed to be customized (for example Msg is encoded in gob).
@@ -26,55 +25,59 @@ func (m ErrorMessage) Msg() interface{} {
 
 // The message the master sends to a newly connected node with its id
 type ReadyMessage struct {
-	Id int
+	Id      int
+	EntryId int
 }
 
 func (m ReadyMessage) Msg() interface{} {
-	return strconv.Itoa(m.Id)
+	return m
 }
 
 func (m ReadyMessage) Type() string {
-	return "ready"
+	return "readyreq"
 }
 
 // The message a node sends to the node it's newly connected to with its id to make authentication easier
-type IdReqMessage struct {
-	Id int
+type GreetingMessage struct {
+	Id   int
+	Data interface{}
 }
 
-func (m IdReqMessage) Msg() interface{} {
-	return strconv.Itoa(m.Id)
+func (m GreetingMessage) Msg() interface{} {
+	return m
 }
 
-func (m IdReqMessage) Type() string {
-	return "idreq"
+func (m GreetingMessage) Type() string {
+	return "greetreq"
 }
 
 // The message the master sends when all nodes when a new node joins
 type NewNodeMessage struct {
-	Id   int    // The Id
-	Addr string // The address to connect to
+	Id   int         // The Id
+	Addr string      // The address to connect to
+	Data interface{} // The new node's data
 }
 
 func (m NewNodeMessage) Msg() interface{} {
 	return m
 }
 
 func (m NewNodeMessage) Type() string {
-	return "newnode"
+	return "newnodereq"
 }
 
-// The message a new node sends to the master with the incoming connections address
-type AddressMessage struct {
+// The message a new node sends to the master with its information
+type IntroduceMessage struct {
 	Addr string
+	Data interface{}
 }
 
-func (m AddressMessage) Msg() interface{} {
+func (m IntroduceMessage) Msg() interface{} {
 	return m.Addr
 }
 
-func (m AddressMessage) Type() string {
-	return "addrreq"
+func (m IntroduceMessage) Type() string {
+	return "introreq"
 }
 
 // Register the message type to gob

node.go

@@ -11,59 +11,57 @@ import (
 
 // The node is the general data type in go-cluster, it resembles a node in the cluster
 type Node struct {
+	Addr    string       // Incoming connections address
 	Ready   bool         // Whether is node is ready for connections
 	Id      int          // This node's ID
-	NextId  int          // (Master Only) Next client's id
+	NextId  int          // Next client's id
 	Message chan Message // The channel to forward messages to
 	Nodes   *sync.Map    // A map that maps other node ids to their connections.
+	Data    interface{}  // Allows the node to have extra information attached to it
 }
 
 func Init() {
 	gob.Register(ReadyMessage{})
 	gob.Register(ErrorMessage{})
-	gob.Register(IdReqMessage{})
+	gob.Register(GreetingMessage{})
 	gob.Register(NewNodeMessage{})
-	gob.Register(AddressMessage{})
+	gob.Register(IntroduceMessage{})
+	gob.Register(Transport{})
 }
 
 // Creates a new master node
 // This new node will introduce other nodes to each other.
-func CreateCluster(address string) *Node {
+func CreateCluster(address string, data interface{}) *Node {
 	node := &Node{
+		Addr:    address,
 		Ready:   true,
 		NextId:  1,
 		Message: make(chan Message),
 		Nodes:   new(sync.Map),
+		Data:    data,
 	}
 	go handleIncoming(address, node)
 	return node
 }
 
 // Creates a new node and connects to the master
-func JoinCluster(address, maddress string) (*Node, error) {
+func JoinCluster(address, maddress string, data interface{}) (*Node, error) {
 	node := &Node{
+		Addr:    address,
 		Ready:   true,
 		Id:      1,
 		NextId:  1,
 		Message: make(chan Message),
 		Nodes:   new(sync.Map),
+		Data:    data,
 	}
-	// TODO add id
 	if conn, err := connect(maddress); err != nil {
 		return nil, err
 	} else if conn == nil {
 		return nil, errors.New("connection cannot be nil, unexpected error occurred")
 	} else {
-		node.Nodes.Store(0, conn)
-		go handleMessages(conn, node, 0)
-		readymsg := <-node.Message
-		node.Id = readymsg.(ReadyMessage).Id
-		node.NextId = node.Id + 1
-		log.Println("Node Intialized! Id:", node.Id, "Address:", address)
 		go handleIncoming(address, node)
-		if err := node.Send(AddressMessage{Addr: address}, 0); err != nil {
-			return nil, err
-		}
+		go handleMessages(conn, node, 0)
 		return node, nil
 	}
 }
@@ -105,7 +103,6 @@ func (n Node) Broadcast(message Message, except ...int) error {
 // Shuts down the node, if the node is the master it will broadcast the new master to all nodes before closing.
 // It accepts a channel that will receive a boolean when the node is shutdown.
 func (n *Node) Close() {
-	// TODO new master broadcast
 	n.Log("Shutting down...")
 	n.Nodes.Range(func(id, conn interface{}) bool {
 		if err := conn.(*Connection).Close(); err != nil {