Merge pull request #98 from iotaledger/feat/glumb

Ports graph plugin from Hornet to GoShimmer

README.md

@@ -69,3 +69,14 @@ To start Shimmer in the background, you can also simply use [Docker Compose](htt
 ```
 docker-compose up -d
 ```
+
+### Install Glumb visualizer
+
+Install both the Glumb visualizer and socket.io client lib within the root folder/where the binary is located:
+```bash
+git clone https://github.com/glumb/IOTAtangle.git
+// only this version seems to be stable
+cd IOTAtangle && git reset --hard 07bba77a296a2d06277cdae56aa963abeeb5f66e 
+cd ../
+git clone https://github.com/socketio/socket.io-client.git
+```

config.json

 {
-  "analysis":{
-    "serveraddress":"ressims.iota.cafe:188",
-    "serverport":0
+  "analysis": {
+    "serveraddress": "ressims.iota.cafe:188",
+    "serverport": 0
   },
-  "autopeering":{
-    "address":"0.0.0.0",
-    "entrynodes":[
+  "autopeering": {
+    "address": "0.0.0.0",
+    "entrynodes": [
       "V8LYtWWcPYYDTTXLeIEFjJEuWlsjDiI0+Pq/Cx9ai6g=@116.202.49.178:14626"
     ],
     "port":14626
   },
-  "database":{
-    "directory":"mainnetdb"
+  "database": {
+    "directory": "mainnetdb"
   },
-  "gossip":{
-    "port":14666
+  "gossip": {
+    "port": 14666
   },
-  "logger":{
-    "Level":"info",
-    "DisableCaller":false,
-    "DisableStacktrace":false,
-    "Encoding":"console",
-    "OutputPaths":[
+  "graph": {
+    "webrootPath": "./IOTAtangle/webroot",
+    "socketioPath": "./socket.io-client/dist/socket.io.js",
+    "domain": "",
+    "host": "127.0.0.1",
+    "port": 8083,
+    "networkName": "GoShimmer"
+  },
+  "logger": {
+    "Level": "info",
+    "DisableCaller": false,
+    "DisableStacktrace": false,
+    "Encoding": "console",
+    "OutputPaths": [
       "shimmer.log"
     ]
   },
-  "node":{
-    "disableplugins":"",
-    "enableplugins":"",
-    "loglevel":0
+  "node": {
+    "disableplugins": "",
+    "enableplugins": [],
+    "loglevel": 0
   }
 }
\ No newline at end of file

go.mod

@@ -10,6 +10,8 @@ require (
 	github.com/gdamore/tcell v1.3.0
 	github.com/go-zeromq/zmq4 v0.7.0
 	github.com/golang/protobuf v1.3.2
+	github.com/googollee/go-engine.io v1.4.3-0.20190924125625-798118fc0dd2
+	github.com/googollee/go-socket.io v1.4.3-0.20191204093753-683f8725b6d0
 	github.com/gorilla/websocket v1.4.1
 	github.com/iotaledger/hive.go v0.0.0-20200110132858-ea86cdb9d91e
 	github.com/iotaledger/iota.go v1.0.0-beta.13

main.go

@@ -11,6 +11,7 @@ import (
 	"github.com/iotaledger/goshimmer/plugins/dashboard"
 	"github.com/iotaledger/goshimmer/plugins/gossip"
 	"github.com/iotaledger/goshimmer/plugins/gracefulshutdown"
+	"github.com/iotaledger/goshimmer/plugins/graph"
 	"github.com/iotaledger/goshimmer/plugins/metrics"
 	"github.com/iotaledger/goshimmer/plugins/statusscreen"
 	statusscreen_tps "github.com/iotaledger/goshimmer/plugins/statusscreen-tps"
@@ -54,9 +55,12 @@ func main() {
 			webapi_spammer.PLUGIN,
 			webapi_send_data.PLUGIN,
 			webapi_tx_request.PLUGIN,
+			webapi_spammer.PLUGIN,
 
 			ui.PLUGIN,
 			webauth.PLUGIN,
+
+			graph.PLUGIN,
 		),
 	)
 }

plugins/graph/graph.go

+package graph
+
+import (
+	"container/ring"
+	"fmt"
+	"strconv"
+	"strings"
+
+	socketio "github.com/googollee/go-socket.io"
+	"github.com/iotaledger/goshimmer/packages/model/value_transaction"
+	"github.com/iotaledger/goshimmer/packages/parameter"
+	"github.com/iotaledger/iota.go/consts"
+
+	"github.com/iotaledger/hive.go/syncutils"
+)
+
+const (
+	TX_BUFFER_SIZE = 1800
+)
+
+var (
+	txRingBuffer *ring.Ring // transactions
+	snRingBuffer *ring.Ring // confirmed transactions
+	msRingBuffer *ring.Ring // Milestones
+
+	broadcastLock    = syncutils.Mutex{}
+	txRingBufferLock = syncutils.Mutex{}
+)
+
+type wsTransaction struct {
+	Hash              string `json:"hash"`
+	Address           string `json:"address"`
+	Value             string `json:"value"`
+	Tag               string `json:"tag"`
+	Timestamp         string `json:"timestamp"`
+	CurrentIndex      string `json:"current_index"`
+	LastIndex         string `json:"last_index"`
+	Bundle            string `json:"bundle_hash"`
+	TrunkTransaction  string `json:"transaction_trunk"`
+	BranchTransaction string `json:"transaction_branch"`
+}
+
+type wsTransactionSn struct {
+	Hash              string `json:"hash"`
+	Address           string `json:"address"`
+	TrunkTransaction  string `json:"transaction_trunk"`
+	BranchTransaction string `json:"transaction_branch"`
+	Bundle            string `json:"bundle"`
+}
+
+type wsConfig struct {
+	NetworkName string `json:"networkName"`
+}
+
+func initRingBuffers() {
+	txRingBuffer = ring.New(TX_BUFFER_SIZE)
+	snRingBuffer = ring.New(TX_BUFFER_SIZE)
+	msRingBuffer = ring.New(20)
+}
+
+func onConnectHandler(s socketio.Conn) error {
+	infoMsg := "Graph client connection established"
+	if s != nil {
+		infoMsg = fmt.Sprintf("%s (ID: %v)", infoMsg, s.ID())
+	}
+	log.Info(infoMsg)
+	socketioServer.JoinRoom("broadcast", s)
+
+	config := &wsConfig{NetworkName: parameter.NodeConfig.GetString("graph.networkName")}
+
+	var initTxs []*wsTransaction
+	txRingBuffer.Do(func(tx interface{}) {
+		if tx != nil {
+			initTxs = append(initTxs, tx.(*wsTransaction))
+		}
+	})
+
+	var initSns []*wsTransactionSn
+	snRingBuffer.Do(func(sn interface{}) {
+		if sn != nil {
+			initSns = append(initSns, sn.(*wsTransactionSn))
+		}
+	})
+
+	var initMs []string
+	msRingBuffer.Do(func(ms interface{}) {
+		if ms != nil {
+			initMs = append(initMs, ms.(string))
+		}
+	})
+
+	s.Emit("config", config)
+	s.Emit("inittx", initTxs)
+	s.Emit("initsn", initSns)
+	s.Emit("initms", initMs)
+	s.Emit("donation", "0")
+	s.Emit("donations", []int{})
+	s.Emit("donation-address", "-")
+
+	return nil
+}
+
+func onErrorHandler(conn socketio.Conn, e error) {
+	errorMsg := "Graph meet error"
+	if e != nil {
+		errorMsg = fmt.Sprintf("%s: %s", errorMsg, e.Error())
+	}
+	log.Error(errorMsg)
+}
+
+func onDisconnectHandler(s socketio.Conn, msg string) {
+	infoMsg := "Graph client connection closed"
+	if s != nil {
+		infoMsg = fmt.Sprintf("%s (ID: %v)", infoMsg, s.ID())
+	}
+	log.Info(fmt.Sprintf("%s: %s", infoMsg, msg))
+	socketioServer.LeaveAllRooms(s)
+}
+
+var emptyTag = strings.Repeat("9", consts.TagTrinarySize/3)
+
+func onNewTx(tx *value_transaction.ValueTransaction) {
+	wsTx := &wsTransaction{
+		Hash:              tx.GetHash(),
+		Address:           tx.GetAddress(),
+		Value:             strconv.FormatInt(tx.GetValue(), 10),
+		Tag:               emptyTag,
+		Timestamp:         strconv.FormatInt(int64(tx.GetTimestamp()), 10),
+		CurrentIndex:      "0",
+		LastIndex:         "0",
+		Bundle:            consts.NullHashTrytes,
+		TrunkTransaction:  tx.GetTrunkTransactionHash(),
+		BranchTransaction: tx.GetBranchTransactionHash(),
+	}
+
+	txRingBufferLock.Lock()
+	txRingBuffer.Value = wsTx
+	txRingBuffer = txRingBuffer.Next()
+	txRingBufferLock.Unlock()
+
+	broadcastLock.Lock()
+	socketioServer.BroadcastToRoom("broadcast", "tx", wsTx)
+	broadcastLock.Unlock()
+}

plugins/graph/parameters.go

+package graph
+
+import (
+	"github.com/iotaledger/goshimmer/packages/parameter"
+)
+
+func init() {
+
+	// "Path to IOTA Tangle Visualiser webroot files"
+	parameter.NodeConfig.SetDefault("graph.webrootPath", "IOTAtangle/webroot")
+
+	// "Path to socket.io.js"
+	parameter.NodeConfig.SetDefault("graph.socketioPath", "socket.io-client/dist/socket.io.js")
+
+	// "Set the domain on which IOTA Tangle Visualiser is served"
+	parameter.NodeConfig.SetDefault("graph.domain", "")
+
+	// "Set the host to which the IOTA Tangle Visualiser listens"
+	parameter.NodeConfig.SetDefault("graph.host", "127.0.0.1")
+
+	// "IOTA Tangle Visualiser webserver port"
+	parameter.NodeConfig.SetDefault("graph.port", 8083)
+
+	// "Name of the network shown in IOTA Tangle Visualiser"
+	parameter.NodeConfig.SetDefault("graph.networkName", "meets HORNET")
+}

plugins/graph/plugin.go

+package graph
+
+import (
+	"fmt"
+	"net/http"
+	"time"
+
+	"github.com/iotaledger/goshimmer/packages/model/value_transaction"
+	"github.com/iotaledger/goshimmer/packages/parameter"
+	"github.com/iotaledger/goshimmer/plugins/tangle"
+	"golang.org/x/net/context"
+
+	engineio "github.com/googollee/go-engine.io"
+	"github.com/googollee/go-engine.io/transport"
+	"github.com/googollee/go-engine.io/transport/polling"
+	"github.com/googollee/go-engine.io/transport/websocket"
+	socketio "github.com/googollee/go-socket.io"
+
+	"github.com/iotaledger/hive.go/daemon"
+	"github.com/iotaledger/hive.go/events"
+	"github.com/iotaledger/hive.go/logger"
+	"github.com/iotaledger/hive.go/node"
+	"github.com/iotaledger/hive.go/workerpool"
+)
+
+var (
+	PLUGIN = node.NewPlugin("Graph", node.Disabled, configure, run)
+
+	log *logger.Logger
+
+	newTxWorkerCount     = 1
+	newTxWorkerQueueSize = 10000
+	newTxWorkerPool      *workerpool.WorkerPool
+
+	server         *http.Server
+	router         *http.ServeMux
+	socketioServer *socketio.Server
+)
+
+func downloadSocketIOHandler(w http.ResponseWriter, r *http.Request) {
+	http.ServeFile(w, r, parameter.NodeConfig.GetString("graph.socketioPath"))
+}
+
+func configureSocketIOServer() error {
+	var err error
+
+	socketioServer, err = socketio.NewServer(&engineio.Options{
+		PingTimeout:  time.Second * 20,
+		PingInterval: time.Second * 5,
+		Transports: []transport.Transport{
+			polling.Default,
+			websocket.Default,
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	socketioServer.OnConnect("/", onConnectHandler)
+	socketioServer.OnError("/", onErrorHandler)
+	socketioServer.OnDisconnect("/", onDisconnectHandler)
+
+	return nil
+}
+
+func configure(plugin *node.Plugin) {
+	log = logger.NewLogger("Graph")
+	initRingBuffers()
+
+	router = http.NewServeMux()
+
+	// socket.io and web server
+	server = &http.Server{
+		Addr:    fmt.Sprintf("%s:%d", parameter.NodeConfig.GetString("graph.host"), parameter.NodeConfig.GetInt("graph.port")),
+		Handler: router,
+	}
+
+	fs := http.FileServer(http.Dir(parameter.NodeConfig.GetString("graph.webrootPath")))
+
+	if err := configureSocketIOServer(); err != nil {
+		log.Panicf("Graph: %v", err.Error())
+	}
+
+	router.Handle("/", fs)
+	router.HandleFunc("/socket.io/socket.io.js", downloadSocketIOHandler)
+	router.Handle("/socket.io/", socketioServer)
+
+	newTxWorkerPool = workerpool.New(func(task workerpool.Task) {
+		onNewTx(task.Param(0).(*value_transaction.ValueTransaction))
+		task.Return(nil)
+	}, workerpool.WorkerCount(newTxWorkerCount), workerpool.QueueSize(newTxWorkerQueueSize))
+}
+
+func run(plugin *node.Plugin) {
+
+	notifyNewTx := events.NewClosure(func(transaction *value_transaction.ValueTransaction) {
+		newTxWorkerPool.TrySubmit(transaction)
+	})
+
+	daemon.BackgroundWorker("Graph[NewTxWorker]", func(shutdownSignal <-chan struct{}) {
+		log.Info("Starting Graph[NewTxWorker] ... done")
+		tangle.Events.TransactionStored.Attach(notifyNewTx)
+		newTxWorkerPool.Start()
+		<-shutdownSignal
+		tangle.Events.TransactionStored.Detach(notifyNewTx)
+		newTxWorkerPool.Stop()
+		log.Info("Stopping Graph[NewTxWorker] ... done")
+	})
+
+	daemon.BackgroundWorker("Graph Webserver", func(shutdownSignal <-chan struct{}) {
+		go socketioServer.Serve()
+
+		go func() {
+			if err := server.ListenAndServe(); err != nil {
+				log.Error(err.Error())
+			}
+		}()
+
+		log.Infof("You can now access IOTA Tangle Visualiser using: http://%s:%d", parameter.NodeConfig.GetString("graph.host"), parameter.NodeConfig.GetInt("graph.port"))
+
+		<-shutdownSignal
+		log.Info("Stopping Graph ...")
+
+		socketioServer.Close()
+
+		ctx, cancel := context.WithTimeout(context.Background(), 0*time.Second)
+		defer cancel()
+
+		_ = server.Shutdown(ctx)
+		log.Info("Stopping Graph ... done")
+	})
+}