Feat: added an analysis server that shows the network

main.go

@@ -2,6 +2,7 @@ package main
 
 import (
     "github.com/iotaledger/goshimmer/packages/node"
+    "github.com/iotaledger/goshimmer/plugins/analysis"
     "github.com/iotaledger/goshimmer/plugins/autopeering"
     "github.com/iotaledger/goshimmer/plugins/cli"
     "github.com/iotaledger/goshimmer/plugins/gracefulshutdown"
@@ -14,5 +15,6 @@ func main() {
         autopeering.PLUGIN,
         statusscreen.PLUGIN,
         gracefulshutdown.PLUGIN,
+        analysis.PLUGIN,
     )
 }

packages/network/events.go

@@ -52,7 +52,7 @@ func (this *dataEvent) Attach(callback DataConsumer) {
     this.callbacks[reflect.ValueOf(callback).Pointer()] = callback
 }
 
-func (this *dataEvent) Detach(callback ErrorConsumer) {
+func (this *dataEvent) Detach(callback DataConsumer) {
     delete(this.callbacks, reflect.ValueOf(callback).Pointer())
 }
 

plugins/analysis/client/parameters.go

+package client
+
+import "github.com/iotaledger/goshimmer/packages/parameter"
+
+var (
+    SERVER_ADDRESS = parameter.AddString("ANALYSIS/SERVER-ADDRESS", "82.165.29.179:188", "tcp server for collecting analysis information")
+)

plugins/analysis/client/plugin.go

+package client
+
+import (
+    "github.com/iotaledger/goshimmer/packages/accountability"
+    "github.com/iotaledger/goshimmer/packages/daemon"
+    "github.com/iotaledger/goshimmer/packages/network"
+    "github.com/iotaledger/goshimmer/packages/node"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/addnode"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/connectnodes"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/ping"
+    "github.com/iotaledger/goshimmer/plugins/autopeering/protocol"
+    "github.com/iotaledger/goshimmer/plugins/autopeering/types/peer"
+    "github.com/iotaledger/goshimmer/plugins/autopeering/types/request"
+    "github.com/iotaledger/goshimmer/plugins/autopeering/types/response"
+    "github.com/iotaledger/goshimmer/plugins/gossip/neighbormanager"
+    "net"
+    "time"
+)
+
+func Run(plugin *node.Plugin) {
+    daemon.BackgroundWorker(func() {
+        shuttingDown := false
+        for !shuttingDown {
+            if conn, err := net.Dial("tcp", *SERVER_ADDRESS.Value); err != nil {
+                plugin.LogDebug("Could not connect to reporting server: " + err.Error())
+            } else {
+                managedConn := network.NewManagedConnection(conn)
+                eventDispatchers := getEventDispatchers(managedConn)
+
+                reportCurrentStatus(eventDispatchers)
+                setupHooks(managedConn, eventDispatchers)
+
+                shuttingDown = keepConnectionAlive(managedConn)
+            }
+        }
+    })
+}
+
+func getEventDispatchers(conn *network.ManagedConnection) *EventDispatchers {
+    return &EventDispatchers{
+        AddNode: func(nodeId []byte) {
+            conn.Write((&addnode.Packet{ NodeId: nodeId }).Marshal())
+        },
+        ConnectNodes: func(sourceId []byte, targetId []byte) {
+            conn.Write((&connectnodes.Packet{ SourceId: sourceId, TargetId: targetId }).Marshal())
+        },
+    }
+}
+
+func reportCurrentStatus(eventDispatchers *EventDispatchers) {
+    eventDispatchers.AddNode(accountability.OWN_ID.Identifier)
+
+    reportChosenNeighbors(eventDispatchers)
+}
+
+func setupHooks(conn *network.ManagedConnection, eventDispatchers *EventDispatchers) {
+    // define hooks ////////////////////////////////////////////////////////////////////////////////////////////////////
+
+    onDiscoverPeer := func(p *peer.Peer) {
+        eventDispatchers.AddNode(p.Identity.Identifier)
+    }
+
+    onIncomingRequestAccepted := func(req *request.Request) {
+        eventDispatchers.ConnectNodes(req.Issuer.Identity.Identifier, accountability.OWN_ID.Identifier)
+    }
+
+    onOutgoingRequestAccepted := func(res *response.Response) {
+        eventDispatchers.ConnectNodes(accountability.OWN_ID.Identifier, res.Issuer.Identity.Identifier)
+    }
+
+    // setup hooks /////////////////////////////////////////////////////////////////////////////////////////////////////
+
+    protocol.Events.DiscoverPeer.Attach(onDiscoverPeer)
+    protocol.Events.IncomingRequestAccepted.Attach(onIncomingRequestAccepted)
+    protocol.Events.OutgoingRequestAccepted.Attach(onOutgoingRequestAccepted)
+
+    // clean up hooks on close /////////////////////////////////////////////////////////////////////////////////////////
+
+    var onClose func()
+    onClose = func() {
+        protocol.Events.DiscoverPeer.Detach(onDiscoverPeer)
+        protocol.Events.IncomingRequestAccepted.Detach(onIncomingRequestAccepted)
+        protocol.Events.OutgoingRequestAccepted.Detach(onOutgoingRequestAccepted)
+
+        conn.Events.Close.Detach(onClose)
+    }
+    conn.Events.Close.Attach(onClose)
+}
+
+func reportChosenNeighbors(dispatchers *EventDispatchers) {
+    for _, chosenNeighbor := range neighbormanager.CHOSEN_NEIGHBORS {
+        dispatchers.AddNode(chosenNeighbor.Identity.Identifier)
+    }
+    for _, chosenNeighbor := range neighbormanager.CHOSEN_NEIGHBORS {
+        dispatchers.ConnectNodes(accountability.OWN_ID.Identifier, chosenNeighbor.Identity.Identifier)
+    }
+}
+
+func keepConnectionAlive(conn *network.ManagedConnection) bool {
+    go conn.Read(make([]byte, 1))
+
+    ticker := time.NewTicker(1 * time.Second)
+    for {
+        select {
+            case <- daemon.ShutdownSignal:
+                return true
+
+            case <- ticker.C:
+                if _, err := conn.Write((&ping.Packet{}).Marshal()); err != nil {
+                    return false
+                }
+        }
+    }
+}

plugins/analysis/client/types.go

+package client
+
+type EventDispatchers struct {
+    AddNode      func(nodeId []byte)
+    ConnectNodes func(sourceId []byte, targetId []byte)
+}

plugins/analysis/plugin.go

+package analysis
+
+import (
+    "github.com/iotaledger/goshimmer/packages/daemon"
+    "github.com/iotaledger/goshimmer/packages/node"
+    "github.com/iotaledger/goshimmer/plugins/analysis/client"
+    "github.com/iotaledger/goshimmer/plugins/analysis/server"
+    "github.com/iotaledger/goshimmer/plugins/analysis/webinterface"
+)
+
+var PLUGIN = node.NewPlugin("Analysis", configure, run)
+
+func configure(plugin *node.Plugin) {
+    if *server.SERVER_PORT.Value != 0 {
+        webinterface.Configure(plugin)
+        server.Configure(plugin)
+
+        server.Events.AddNode.Attach(func(nodeId string) {
+            
+        })
+
+        daemon.Events.Shutdown.Attach(func() {
+            server.Shutdown(plugin)
+        })
+    }
+}
+
+func run(plugin *node.Plugin) {
+    if *server.SERVER_PORT.Value != 0 {
+        webinterface.Run(plugin)
+        server.Run(plugin)
+    } else {
+        plugin.Node.LogSuccess("Node", "Starting Plugin: Analysis ... server is disabled (server-port is 0)")
+    }
+
+    if *client.SERVER_ADDRESS.Value != "" {
+        client.Run(plugin)
+    } else {
+        plugin.Node.LogSuccess("Node", "Starting Plugin: Analysis ... client is disabled (server-address is empty)")
+    }
+}

plugins/analysis/server/constants.go

+package server
+
+import (
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/addnode"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/connectnodes"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/disconnectnodes"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/ping"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/removenode"
+    "time"
+)
+
+const (
+    IDLE_TIMEOUT = 5 * time.Second
+
+    STATE_INITIAL          = byte(255)
+    STATE_INITIAL_ADDNODE  = byte(254)
+    STATE_CONSECUTIVE      = byte(253)
+    STATE_PING             = ping.MARSHALLED_PACKET_HEADER
+    STATE_ADD_NODE         = addnode.MARSHALLED_PACKET_HEADER
+    STATE_REMOVE_NODE      = removenode.MARSHALLED_PACKET_HEADER
+    STATE_CONNECT_NODES    = connectnodes.MARSHALLED_PACKET_HEADER
+    STATE_DISCONNECT_NODES = disconnectnodes.MARSHALLED_PACKET_HEADER
+)

plugins/analysis/server/events.go

+package server
+
+import (
+    "reflect"
+)
+
+var Events = &pluginEvents{
+    AddNode:         &nodeIdEvent{make(map[uintptr]StringConsumer)},
+    RemoveNode:      &nodeIdEvent{make(map[uintptr]StringConsumer)},
+    ConnectNodes:    &nodeIdsEvent{make(map[uintptr]StringStringConsumer)},
+    DisconnectNodes: &nodeIdsEvent{make(map[uintptr]StringStringConsumer)},
+    NodeOnline:      &nodeIdEvent{make(map[uintptr]StringConsumer)},
+    NodeOffline:     &nodeIdEvent{make(map[uintptr]StringConsumer)},
+    Error:           &errorEvent{make(map[uintptr]ErrorConsumer)},
+}
+
+type pluginEvents struct {
+    AddNode         *nodeIdEvent
+    RemoveNode      *nodeIdEvent
+    ConnectNodes    *nodeIdsEvent
+    DisconnectNodes *nodeIdsEvent
+    NodeOnline      *nodeIdEvent
+    NodeOffline     *nodeIdEvent
+    Error           *errorEvent
+}
+
+type nodeIdEvent struct {
+    callbacks map[uintptr]StringConsumer
+}
+
+func (this *nodeIdEvent) Attach(callback StringConsumer) {
+    this.callbacks[reflect.ValueOf(callback).Pointer()] = callback
+}
+
+func (this *nodeIdEvent) Detach(callback StringConsumer) {
+    delete(this.callbacks, reflect.ValueOf(callback).Pointer())
+}
+
+func (this *nodeIdEvent) Trigger(nodeId string) {
+    for _, callback := range this.callbacks {
+        callback(nodeId)
+    }
+}
+
+type nodeIdsEvent struct {
+    callbacks map[uintptr]StringStringConsumer
+}
+
+func (this *nodeIdsEvent) Attach(callback StringStringConsumer) {
+    this.callbacks[reflect.ValueOf(callback).Pointer()] = callback
+}
+
+func (this *nodeIdsEvent) Detach(callback StringStringConsumer) {
+    delete(this.callbacks, reflect.ValueOf(callback).Pointer())
+}
+
+func (this *nodeIdsEvent) Trigger(sourceId string, targetId string) {
+    for _, callback := range this.callbacks {
+        callback(sourceId, targetId)
+    }
+}
+
+type errorEvent struct {
+    callbacks map[uintptr]ErrorConsumer
+}
+
+func (this *errorEvent) Attach(callback ErrorConsumer) {
+    this.callbacks[reflect.ValueOf(callback).Pointer()] = callback
+}
+
+func (this *errorEvent) Detach(callback ErrorConsumer) {
+    delete(this.callbacks, reflect.ValueOf(callback).Pointer())
+}
+
+func (this *errorEvent) Trigger(err error) {
+    for _, callback := range this.callbacks {
+        callback(err)
+    }
+}
+
+type ErrorConsumer = func(err error)
+
+type StringConsumer = func(str string)
+
+type StringStringConsumer = func(sourceId string, targetId string)

plugins/analysis/server/parameters.go

+package server
+
+import "github.com/iotaledger/goshimmer/packages/parameter"
+
+var (
+    SERVER_PORT = parameter.AddInt("ANALYSIS/SERVER-PORT", 0, "tcp port for incoming analysis packets")
+)

plugins/analysis/server/plugin.go

+package server
+
+import (
+    "encoding/hex"
+    "github.com/iotaledger/goshimmer/packages/daemon"
+    "github.com/iotaledger/goshimmer/packages/network"
+    "github.com/iotaledger/goshimmer/packages/network/tcp"
+    "github.com/iotaledger/goshimmer/packages/node"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/addnode"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/connectnodes"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/disconnectnodes"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/ping"
+    "github.com/iotaledger/goshimmer/plugins/analysis/types/removenode"
+    "github.com/pkg/errors"
+    "math"
+    "strconv"
+)
+
+var server *tcp.Server
+
+func Configure(plugin *node.Plugin) {
+    server = tcp.NewServer()
+
+    server.Events.Connect.Attach(HandleConnection)
+    server.Events.Error.Attach(func(err error) {
+        plugin.LogFailure("error in server: " + err.Error())
+    })
+    server.Events.Start.Attach(func() {
+        plugin.LogSuccess("Starting Server (port " + strconv.Itoa(*SERVER_PORT.Value) + ") ... done")
+    })
+    server.Events.Shutdown.Attach(func() {
+        plugin.LogSuccess("Stopping Server ... done")
+    })
+}
+
+func Run(plugin *node.Plugin) {
+    daemon.BackgroundWorker(func() {
+        plugin.LogInfo("Starting Server (port " + strconv.Itoa(*SERVER_PORT.Value) + ") ...")
+
+        server.Listen(*SERVER_PORT.Value)
+    })
+}
+
+func Shutdown(plugin *node.Plugin) {
+    plugin.LogInfo("Stopping Server ...")
+
+    server.Shutdown()
+}
+
+func HandleConnection(conn *network.ManagedConnection) {
+    conn.SetTimeout(IDLE_TIMEOUT)
+
+    var connectionState = STATE_INITIAL
+    var receiveBuffer []byte
+    var offset int
+    var connectedNodeId string
+
+    var onReceiveData func(data []byte)
+    var onDisconnect func()
+
+    onReceiveData = func(data []byte) {
+        processIncomingPacket(&connectionState, &receiveBuffer, conn, data, &offset, &connectedNodeId)
+    }
+    onDisconnect = func() {
+        Events.NodeOffline.Trigger(connectedNodeId)
+
+        conn.Events.ReceiveData.Detach(onReceiveData)
+        conn.Events.Close.Detach(onDisconnect)
+    }
+
+    conn.Events.ReceiveData.Attach(onReceiveData)
+    conn.Events.Close.Attach(onDisconnect)
+
+    maxPacketsSize := getMaxPacketSize(
+        ping.MARSHALLED_TOTAL_SIZE,
+        addnode.MARSHALLED_TOTAL_SIZE,
+        removenode.MARSHALLED_TOTAL_SIZE,
+        connectnodes.MARSHALLED_TOTAL_SIZE,
+        disconnectnodes.MARSHALLED_PACKET_HEADER,
+    )
+
+    go conn.Read(make([]byte, maxPacketsSize))
+}
+
+func getMaxPacketSize(packetSizes ...int) int {
+    maxPacketSize := 0
+
+    for _, packetSize := range packetSizes {
+        if packetSize > maxPacketSize {
+            maxPacketSize = packetSize
+        }
+    }
+
+    return maxPacketSize
+}
+
+func processIncomingPacket(connectionState *byte, receiveBuffer *[]byte, conn *network.ManagedConnection, data []byte, offset *int, connectedNodeId *string) {
+    firstPackage := *connectionState == STATE_INITIAL
+
+    if firstPackage || *connectionState == STATE_CONSECUTIVE {
+        var err error
+        if *connectionState, *receiveBuffer, err = parsePackageHeader(data); err != nil {
+            Events.Error.Trigger(err)
+
+            conn.Close()
+
+            return
+        }
+
+        *offset = 0
+
+        switch *connectionState {
+            case STATE_ADD_NODE:
+                *receiveBuffer = make([]byte, addnode.MARSHALLED_TOTAL_SIZE)
+
+            case STATE_PING:
+                *receiveBuffer = make([]byte, ping.MARSHALLED_TOTAL_SIZE)
+
+            case STATE_CONNECT_NODES:
+                *receiveBuffer = make([]byte, connectnodes.MARSHALLED_TOTAL_SIZE)
+
+            case STATE_REMOVE_NODE:
+                *receiveBuffer = make([]byte, removenode.MARSHALLED_TOTAL_SIZE)
+        }
+    }
+
+    if firstPackage {
+        if *connectionState != STATE_ADD_NODE {
+            Events.Error.Trigger(errors.New("expected initial add node package"))
+        } else {
+            *connectionState = STATE_INITIAL_ADDNODE
+        }
+    }
+
+    switch *connectionState {
+        case STATE_INITIAL_ADDNODE:
+            processIncomingAddNodePacket(connectionState, receiveBuffer, conn, data, offset, connectedNodeId)
+
+        case STATE_ADD_NODE:
+            processIncomingAddNodePacket(connectionState, receiveBuffer, conn, data, offset, connectedNodeId)
+
+        case STATE_PING:
+            processIncomingPingPacket(connectionState, receiveBuffer, conn, data, offset, connectedNodeId)
+
+        case STATE_CONNECT_NODES:
+            processIncomingConnectNodesPacket(connectionState, receiveBuffer, conn, data, offset, connectedNodeId)
+
+        case STATE_REMOVE_NODE:
+            processIncomingAddNodePacket(connectionState, receiveBuffer, conn, data, offset, connectedNodeId)
+    }
+}
+
+func parsePackageHeader(data []byte) (ConnectionState, []byte, error) {
+    var connectionState ConnectionState
+    var receiveBuffer []byte
+
+    switch data[0] {
+        case ping.MARSHALLED_PACKET_HEADER:
+            receiveBuffer = make([]byte, ping.MARSHALLED_TOTAL_SIZE)
+
+            connectionState = STATE_PING
+
+        case addnode.MARSHALLED_PACKET_HEADER:
+            receiveBuffer = make([]byte, addnode.MARSHALLED_TOTAL_SIZE)
+
+            connectionState = STATE_ADD_NODE
+
+        case connectnodes.MARSHALLED_PACKET_HEADER:
+            receiveBuffer = make([]byte, connectnodes.MARSHALLED_TOTAL_SIZE)
+
+            connectionState = STATE_CONNECT_NODES
+
+        case removenode.MARSHALLED_PACKET_HEADER:
+            receiveBuffer = make([]byte, removenode.MARSHALLED_TOTAL_SIZE)
+
+            connectionState = STATE_REMOVE_NODE
+
+        default:
+            return 0, nil, errors.New("invalid package header")
+    }
+
+    return connectionState, receiveBuffer, nil
+}
+
+func processIncomingAddNodePacket(connectionState *byte, receiveBuffer *[]byte, conn *network.ManagedConnection, data []byte, offset *int, connectedNodeId *string) {
+    remainingCapacity := int(math.Min(float64(addnode.MARSHALLED_TOTAL_SIZE- *offset), float64(len(data))))
+
+    copy((*receiveBuffer)[*offset:], data[:remainingCapacity])
+
+    if *offset + len(data) < addnode.MARSHALLED_TOTAL_SIZE {
+        *offset += len(data)
+    } else {
+        if addNodePacket, err := addnode.Unmarshal(*receiveBuffer); err != nil {
+            Events.Error.Trigger(err)
+
+            conn.Close()
+
+            return
+        } else {
+            nodeId := hex.EncodeToString(addNodePacket.NodeId)
+
+            Events.AddNode.Trigger(nodeId)
+
+            if *connectionState == STATE_INITIAL_ADDNODE {
+                *connectedNodeId = nodeId
+
+                Events.NodeOnline.Trigger(nodeId)
+            }
+        }
+
+        *connectionState = STATE_CONSECUTIVE
+
+        if *offset + len(data) > addnode.MARSHALLED_TOTAL_SIZE {
+            processIncomingPacket(connectionState, receiveBuffer, conn, data[remainingCapacity:], offset, connectedNodeId)
+        }
+    }
+}
+
+func processIncomingPingPacket(connectionState *byte, receiveBuffer *[]byte, conn *network.ManagedConnection, data []byte, offset *int, connectedNodeId *string) {
+    remainingCapacity := int(math.Min(float64(ping.MARSHALLED_TOTAL_SIZE- *offset), float64(len(data))))
+
+    copy((*receiveBuffer)[*offset:], data[:remainingCapacity])
+
+    if *offset + len(data) < ping.MARSHALLED_TOTAL_SIZE {
+        *offset += len(data)
+    } else {
+        if _, err := ping.Unmarshal(*receiveBuffer); err != nil {
+            Events.Error.Trigger(err)
+
+            conn.Close()
+
+            return
+        }
+
+        *connectionState = STATE_CONSECUTIVE
+
+        if *offset + len(data) > ping.MARSHALLED_TOTAL_SIZE {
+            processIncomingPacket(connectionState, receiveBuffer, conn, data[remainingCapacity:], offset, connectedNodeId)
+        }
+    }
+}
+
+func processIncomingConnectNodesPacket(connectionState *byte, receiveBuffer *[]byte, conn *network.ManagedConnection, data []byte, offset *int, connectedNodeId *string) {
+    remainingCapacity := int(math.Min(float64(connectnodes.MARSHALLED_TOTAL_SIZE- *offset), float64(len(data))))
+
+    copy((*receiveBuffer)[*offset:], data[:remainingCapacity])
+
+    if *offset + len(data) < connectnodes.MARSHALLED_TOTAL_SIZE {
+        *offset += len(data)
+    } else {
+        if connectNodesPacket, err := connectnodes.Unmarshal(*receiveBuffer); err != nil {
+            Events.Error.Trigger(err)
+
+            conn.Close()
+
+            return
+        } else {
+            sourceNodeId := hex.EncodeToString(connectNodesPacket.SourceId)
+            targetNodeId := hex.EncodeToString(connectNodesPacket.TargetId)
+
+            Events.ConnectNodes.Trigger(sourceNodeId, targetNodeId)
+        }
+
+        *connectionState = STATE_CONSECUTIVE
+
+        if *offset + len(data) > connectnodes.MARSHALLED_TOTAL_SIZE {
+            processIncomingPacket(connectionState, receiveBuffer, conn, data[remainingCapacity:], offset, connectedNodeId)
+        }
+    }
+}

plugins/analysis/server/types.go

+package server
+
+type ConnectionState = byte

plugins/analysis/types/addnode/constants.go

+package addnode
+
+const (
+    MARSHALLED_PACKET_HEADER = 0x01
+
+    MARSHALLED_PACKET_HEADER_START = 0
+    MARSHALLED_PACKET_HEADER_SIZE  = 1
+    MARSHALLED_PACKET_HEADER_END   = MARSHALLED_PACKET_HEADER_START + MARSHALLED_PACKET_HEADER_SIZE
+
+    MARSHALLED_ID_START = MARSHALLED_PACKET_HEADER_END
+    MARSHALLED_ID_SIZE  = 20
+    MARSHALLED_ID_END   = MARSHALLED_ID_START + MARSHALLED_ID_SIZE
+
+    MARSHALLED_TOTAL_SIZE = MARSHALLED_ID_END
+)

plugins/analysis/types/addnode/packet.go

+package addnode
+
+import "github.com/pkg/errors"
+
+type Packet struct {
+    NodeId []byte
+}
+
+func Unmarshal(data []byte) (*Packet, error) {
+    if len(data) < MARSHALLED_TOTAL_SIZE || data[0] != MARSHALLED_PACKET_HEADER {
+        return nil, errors.New("malformed add node packet")
+    }
+
+    unmarshalledPackage := &Packet{
+        NodeId: make([]byte, MARSHALLED_ID_SIZE),
+    }
+
+    copy(unmarshalledPackage.NodeId, data[MARSHALLED_ID_START:MARSHALLED_ID_END])
+
+    return unmarshalledPackage, nil
+}
+
+func (packet *Packet) Marshal() []byte {
+    marshalledPackage := make([]byte, MARSHALLED_TOTAL_SIZE)
+
+    marshalledPackage[MARSHALLED_PACKET_HEADER_START] = MARSHALLED_PACKET_HEADER
+    copy(marshalledPackage[MARSHALLED_ID_START:MARSHALLED_ID_END], packet.NodeId[:MARSHALLED_ID_SIZE])
+
+    return marshalledPackage
+}

plugins/analysis/types/connectnodes/constants.go

+package connectnodes
+
+const (
+    MARSHALLED_PACKET_HEADER = 0x03
+
+    MARSHALLED_PACKET_HEADER_START = 0
+    MARSHALLED_PACKET_HEADER_SIZE  = 1
+    MARSHALLED_PACKET_HEADER_END   = MARSHALLED_PACKET_HEADER_START + MARSHALLED_PACKET_HEADER_SIZE
+
+    MARSHALLED_SOURCE_ID_START = MARSHALLED_PACKET_HEADER_END
+    MARSHALLED_SOURCE_ID_SIZE  = 20
+    MARSHALLED_SOURCE_ID_END   = MARSHALLED_SOURCE_ID_START + MARSHALLED_SOURCE_ID_SIZE
+
+    MARSHALLED_TARGET_ID_START = MARSHALLED_SOURCE_ID_END
+    MARSHALLED_TARGET_ID_SIZE  = 20
+    MARSHALLED_TARGET_ID_END   = MARSHALLED_TARGET_ID_START + MARSHALLED_TARGET_ID_SIZE
+
+    MARSHALLED_TOTAL_SIZE = MARSHALLED_TARGET_ID_END
+)

plugins/analysis/types/connectnodes/packet.go

+package connectnodes
+
+import "github.com/pkg/errors"
+
+type Packet struct {
+    SourceId []byte
+    TargetId []byte
+}
+
+func Unmarshal(data []byte) (*Packet, error) {
+    if len(data) < MARSHALLED_TOTAL_SIZE || data[0] != MARSHALLED_PACKET_HEADER {
+        return nil, errors.New("malformed connect nodes packet")
+    }
+
+    unmarshalledPackage := &Packet{
+        SourceId: make([]byte, MARSHALLED_SOURCE_ID_SIZE),
+        TargetId: make([]byte, MARSHALLED_TARGET_ID_SIZE),
+    }
+
+    copy(unmarshalledPackage.SourceId, data[MARSHALLED_SOURCE_ID_START:MARSHALLED_SOURCE_ID_END])
+    copy(unmarshalledPackage.TargetId, data[MARSHALLED_TARGET_ID_START:MARSHALLED_TARGET_ID_END])
+
+    return unmarshalledPackage, nil
+}
+
+func (packet *Packet) Marshal() []byte {
+    marshalledPackage := make([]byte, MARSHALLED_TOTAL_SIZE)
+
+    marshalledPackage[MARSHALLED_PACKET_HEADER_START] = MARSHALLED_PACKET_HEADER
+    copy(marshalledPackage[MARSHALLED_SOURCE_ID_START:MARSHALLED_SOURCE_ID_END], packet.SourceId[:MARSHALLED_SOURCE_ID_SIZE])
+    copy(marshalledPackage[MARSHALLED_TARGET_ID_START:MARSHALLED_TARGET_ID_END], packet.TargetId[:MARSHALLED_TARGET_ID_SIZE])
+
+    return marshalledPackage
+}

plugins/analysis/types/disconnectnodes/constants.go

+package disconnectnodes
+
+const (
+    MARSHALLED_PACKET_HEADER = 0x04
+
+    MARSHALLED_PACKET_HEADER_START = 0
+    MARSHALLED_PACKET_HEADER_SIZE  = 1
+    MARSHALLED_PACKET_HEADER_END   = MARSHALLED_PACKET_HEADER_START + MARSHALLED_PACKET_HEADER_SIZE
+
+    MARSHALLED_SOURCE_ID_START = MARSHALLED_PACKET_HEADER_END
+    MARSHALLED_SOURCE_ID_SIZE  = 20
+    MARSHALLED_SOURCE_ID_END   = MARSHALLED_SOURCE_ID_START + MARSHALLED_SOURCE_ID_SIZE
+
+    MARSHALLED_TARGET_ID_START = MARSHALLED_SOURCE_ID_END
+    MARSHALLED_TARGET_ID_SIZE  = 20
+    MARSHALLED_TARGET_ID_END   = MARSHALLED_TARGET_ID_START + MARSHALLED_TARGET_ID_SIZE
+
+    MARSHALLED_TOTAL_SIZE = MARSHALLED_TARGET_ID_END
+)

plugins/analysis/types/disconnectnodes/packet.go

+package disconnectnodes
+
+import "github.com/pkg/errors"
+
+type Packet struct {
+    SourceId []byte
+    TargetId []byte
+}
+
+func Unmarshal(data []byte) (*Packet, error) {
+    if len(data) < MARSHALLED_TOTAL_SIZE || data[0] != MARSHALLED_PACKET_HEADER {
+        return nil, errors.New("malformed disconnect nodes packet")
+    }
+
+    unmarshalledPackage := &Packet{
+        SourceId: make([]byte, MARSHALLED_SOURCE_ID_SIZE),
+        TargetId: make([]byte, MARSHALLED_TARGET_ID_SIZE),
+    }
+
+    copy(unmarshalledPackage.SourceId, data[MARSHALLED_SOURCE_ID_START:MARSHALLED_SOURCE_ID_END])
+    copy(unmarshalledPackage.TargetId, data[MARSHALLED_TARGET_ID_START:MARSHALLED_TARGET_ID_END])
+
+    return unmarshalledPackage, nil
+}
+
+func (packet *Packet) Marshal() []byte {
+    marshalledPackage := make([]byte, MARSHALLED_TOTAL_SIZE)
+
+    marshalledPackage[MARSHALLED_PACKET_HEADER_START] = MARSHALLED_PACKET_HEADER
+    copy(marshalledPackage[MARSHALLED_SOURCE_ID_START:MARSHALLED_SOURCE_ID_END], packet.SourceId[:MARSHALLED_SOURCE_ID_SIZE])
+    copy(marshalledPackage[MARSHALLED_TARGET_ID_START:MARSHALLED_TARGET_ID_END], packet.TargetId[:MARSHALLED_TARGET_ID_SIZE])
+
+    return marshalledPackage
+}

plugins/analysis/types/ping/constants.go

+package ping
+
+const (
+    MARSHALLED_PACKET_HEADER = 0x00
+
+    MARSHALLED_PACKET_HEADER_START = 0
+    MARSHALLED_PACKET_HEADER_SIZE  = 1
+    MARSHALLED_PACKET_HEADER_END   = MARSHALLED_PACKET_HEADER_START + MARSHALLED_PACKET_HEADER_SIZE
+
+    MARSHALLED_TOTAL_SIZE = MARSHALLED_PACKET_HEADER_END
+)

plugins/analysis/types/ping/packet.go

+package ping
+
+import "github.com/pkg/errors"
+
+type Packet struct {}
+
+func Unmarshal(data []byte) (*Packet, error) {
+    if len(data) < MARSHALLED_TOTAL_SIZE || data[MARSHALLED_PACKET_HEADER_START] != MARSHALLED_PACKET_HEADER {
+        return nil, errors.New("malformed ping packet")
+    }
+
+    unmarshalledPacket := &Packet{}
+
+    return unmarshalledPacket, nil
+}
+
+func (packet *Packet) Marshal() []byte {
+    marshalledPackage := make([]byte, MARSHALLED_TOTAL_SIZE)
+
+    marshalledPackage[MARSHALLED_PACKET_HEADER_START] = MARSHALLED_PACKET_HEADER
+
+    return marshalledPackage
+}
\ No newline at end of file

plugins/analysis/types/removenode/constants.go

+package removenode
+
+const (
+    MARSHALLED_PACKET_HEADER = 0x02
+
+    MARSHALLED_PACKET_HEADER_START = 0
+    MARSHALLED_PACKET_HEADER_SIZE  = 1
+    MARSHALLED_PACKET_HEADER_END   = MARSHALLED_PACKET_HEADER_START + MARSHALLED_PACKET_HEADER_SIZE
+
+    MARSHALLED_ID_START = MARSHALLED_PACKET_HEADER_END
+    MARSHALLED_ID_SIZE  = 20
+    MARSHALLED_ID_END   = MARSHALLED_ID_START + MARSHALLED_ID_SIZE
+
+    MARSHALLED_TOTAL_SIZE = MARSHALLED_ID_END
+)