Added websocket connection parameter to messaging functions to prevent panic: runtime error: invalid memory address or nil pointer dereference

Author: nathansenn

api/handlers.go

@@ -222,7 +222,7 @@ func handleMessaging(c *gin.Context) {
 			fmt.Println("Error encoding JSON:", err)
 			return
 		}
-		go messaging.HandleMessage(string(jsonData))
+		go messaging.HandleMessage(string(jsonData), ws)
 	}
 }
 

api/validation.go

@@ -77,7 +77,7 @@ func performPingPong(ctx context.Context, salt string, name string, conn *websoc
 			if messaging.Ping[salt] {
 				return nil
 			}
-			messaging.SendPing(salt, name)
+			messaging.SendPing(salt, name, conn)
 			attempts++
 			if attempts > maxPingAttempts {
 				return sendWsResponseAndHandleError("Connection Error", "Could not connect to peer, try again", "0", conn, errors.New("connection error"))

connection/connection.go

@@ -129,7 +129,7 @@ func StartWsClient(name string) {
 					isConnected = false
 					continue
 				}
-				go messaging.HandleMessage(string(message))
+				go messaging.HandleMessage(string(message), localdata.WsValidators[name])
 				//fmt.Println("Client recv: ", string(message))
 			} else {
 				log.Println("Connection is not established.")

messaging/cids.go

@@ -12,7 +12,7 @@ import (
 	"strings"
 )
 
-func RequestCIDS(req Request) {
+func RequestCIDS(req Request, ws *websocket.Conn) {
 	fmt.Println("Requesting CIDS")
 	data := map[string]string{
 		"type": "RequestCIDS",
@@ -27,7 +27,6 @@ func RequestCIDS(req Request) {
 	if localdata.WsPeers[req.User] == req.User && localdata.NodeType == 1 {
 		WsMutex.Lock()
 		fmt.Println("Locking wsMutex")
-		ws := localdata.WsClients[req.User]
 		ws.WriteMessage(websocket.TextMessage, jsonData)
 		fmt.Println("Sent RequestCIDS to client")
 		WsMutex.Unlock()
@@ -40,7 +39,7 @@ func RequestCIDS(req Request) {
 	}
 
 }
-func SendCIDS(name string) {
+func SendCIDS(name string, ws *websocket.Conn) {
 	allPins, _ := ipfs.Shell.Pins()
 	fmt.Println("Fetched pins")
 	NewPins := make([]string, 0)
@@ -76,8 +75,7 @@ func SendCIDS(name string) {
 		}
 
 		if localdata.UseWS == true && localdata.NodeType == 2 {
-			ws := localdata.WsValidators[name]
-			fmt.Println("Sending CIDS to validation node")
+			fmt.Println("Sending CIDS to validation node", name)
 			WsMutex.Lock()
 			ws.WriteJSON(data)
 			WsMutex.Unlock()

messaging/messaging.go

@@ -50,7 +50,7 @@ type PinMap map[string]PinType
 
 // HandleMessage
 // This is the function that handles the messages from the pubsub
-func HandleMessage(message string) {
+func HandleMessage(message string, ws *websocket.Conn) {
 	fmt.Println("Message received:", message)
 	// JSON decode message
 	req := Request{}
@@ -70,7 +70,7 @@ func HandleMessage(message string) {
 			fmt.Println("User: " + req.User)
 			fmt.Println("CID: " + req.CID)
 			fmt.Println("Pins: " + req.Pins)
-			go HandleProofOfAccess(req)
+			go HandleProofOfAccess(req, ws)
 		}
 
 	}
@@ -79,7 +79,7 @@ func HandleMessage(message string) {
 	if nodeType == 2 {
 		if req.Type == TypeRequestProof {
 			fmt.Println("Request for proof received")
-			go HandleRequestProof(req)
+			go HandleRequestProof(req, ws)
 		}
 		if req.Type == TypePingPongPong {
 			validatorName := req.User
@@ -95,25 +95,25 @@ func HandleMessage(message string) {
 	}
 	if req.Type == TypePingPongPing {
 		fmt.Println("PingPongPing received")
-		PingPongPong(req, req.Hash, req.User)
+		PingPongPong(req, ws)
 		nodeStatus := localdata.NodesStatus[req.User]
 		nodes := Nodes[req.User]
 		fmt.Println("Node Status: " + nodeStatus)
 		fmt.Println("Nodes: " + fmt.Sprint(nodes))
 		if nodeType == 1 && !nodes && nodeStatus != "Synced" {
 			fmt.Println("syncing: " + req.User)
-			go SyncNode(req)
+			go SyncNode(req, ws)
 		}
 
 	}
 	if req.Type == "RequestCIDS" {
 		fmt.Println("RequestCIDS received")
-		go SendCIDS(req.User)
+		go SendCIDS(req.User, ws)
 
 	}
 	if req.Type == "SendCIDS" {
 		fmt.Println("SendCIDS received")
-		go SyncNode(req)
+		go SyncNode(req, ws)
 
 	}
 	if req.Type == "Syncing" {
@@ -148,14 +148,15 @@ func PubsubHandler(ctx context.Context) {
 					log.Error("Error reading from pubsub: ", err)
 					continue
 				}
-				HandleMessage(msg)
+				var ws *websocket.Conn
+				HandleMessage(msg, ws)
 			}
 		}
 	} else {
 		time.Sleep(1 * time.Second)
 	}
 }
-func SendPing(hash string, user string) {
+func SendPing(hash string, user string, ws *websocket.Conn) {
 	fmt.Println("Sending Ping")
 	data := map[string]string{
 		"type": TypePingPongPing,
@@ -169,7 +170,6 @@ func SendPing(hash string, user string) {
 	}
 	localdata.PingTime[user] = time.Now()
 	if localdata.WsPeers[user] == user && localdata.NodeType == 1 {
-		ws := localdata.WsClients[user]
 		WsMutex.Lock()
 		ws.WriteMessage(websocket.TextMessage, jsonData)
 		WsMutex.Unlock()
@@ -181,11 +181,11 @@ func SendPing(hash string, user string) {
 		pubsub.Publish(string(jsonData), user)
 	}
 }
-func PingPongPong(req Request, hash string, user string) {
+func PingPongPong(req Request, ws *websocket.Conn) {
 	fmt.Println("Sending PingPongPong")
 	data := map[string]string{
 		"type": TypePingPongPong,
-		"hash": hash,
+		"hash": req.Hash,
 		"user": localdata.GetNodeName(),
 	}
 	jsonData, err := json.Marshal(data)
@@ -194,7 +194,6 @@ func PingPongPong(req Request, hash string, user string) {
 		return
 	}
 	if localdata.WsPeers[req.User] == req.User && localdata.NodeType == 1 {
-		ws := localdata.WsClients[req.User]
 		fmt.Println("Sending PingPongPong to client")
 		localdata.Lock.Lock()
 		localdata.PeerLastActive[req.User] = time.Now()
@@ -209,6 +208,6 @@ func PingPongPong(req Request, hash string, user string) {
 		localdata.WsValidators[req.User].WriteMessage(websocket.TextMessage, jsonData)
 		WsMutex.Unlock()
 	} else {
-		pubsub.Publish(string(jsonData), user)
+		pubsub.Publish(string(jsonData), req.User)
 	}
 }

messaging/proofs.go

@@ -14,22 +14,22 @@ import (
 
 // HandleRequestProof
 // This is the function that handles the request for proof from the validation node
-func HandleRequestProof(req Request) {
+func HandleRequestProof(req Request, ws *websocket.Conn) {
 	CID := req.CID
 	hash := req.Hash
 	if ipfs.IsPinnedInDB(CID) == true {
 		fmt.Println("Sending proof of access to validation node")
 		validationHash := validation.CreatProofHash(hash, CID)
-		SendProof(req, validationHash, hash, localdata.NodeName)
+		SendProof(req, validationHash, hash, localdata.NodeName, ws)
 	} else {
 		fmt.Println("Pin not found")
-		SendProof(req, "NA", hash, localdata.NodeName)
+		SendProof(req, "NA", hash, localdata.NodeName, ws)
 	}
 }
 
 // HandleProofOfAccess
 // This is the function that handles the proof of access response from the storage node
-func HandleProofOfAccess(req Request) {
+func HandleProofOfAccess(req Request, ws *websocket.Conn) {
 	fmt.Println("Handling proof of access response from storage node")
 	// Get the start time from the seed
 	start := localdata.GetTime(req.Seed)
@@ -82,12 +82,12 @@ func HandleProofOfAccess(req Request) {
 
 // SendProof
 // This is the function that sends the proof of access to the validation node
-func SendProof(req Request, hash string, seed string, user string) {
+func SendProof(req Request, validationHash string, salt string, user string, ws *websocket.Conn) {
 	fmt.Println("Sending proof of access to validation node")
 	data := map[string]string{
 		"type": TypeProofOfAccess,
-		"hash": hash,
-		"seed": seed,
+		"hash": validationHash,
+		"seed": salt,
 		"user": user,
 	}
 	jsonData, err := json.Marshal(data)
@@ -98,12 +98,10 @@ func SendProof(req Request, hash string, seed string, user string) {
 	wsPeers := localdata.WsPeers[req.User]
 	nodeType := localdata.NodeType
 	if wsPeers == req.User && nodeType == 1 {
-		ws := localdata.WsClients[req.User]
 		WsMutex.Lock()
 		ws.WriteMessage(websocket.TextMessage, jsonData)
 		WsMutex.Unlock()
 	} else if localdata.UseWS == true && localdata.NodeType == 2 {
-		ws := localdata.WsValidators[req.User]
 		WsMutex.Lock()
 		ws.WriteMessage(websocket.TextMessage, jsonData)
 		WsMutex.Unlock()

messaging/syncing.go

@@ -8,7 +8,7 @@ import (
 	"proofofaccess/pubsub"
 )
 
-func SendSyncing(req Request) {
+func SendSyncing(req Request, ws *websocket.Conn) {
 	localdata.Synced = true
 	data := map[string]string{
 		"type": "Syncing",
@@ -21,20 +21,18 @@ func SendSyncing(req Request) {
 		return
 	}
 	if localdata.WsPeers[req.User] == req.User && localdata.NodeType == 1 {
-		ws := localdata.WsClients[req.User]
 		WsMutex.Lock()
 		ws.WriteMessage(websocket.TextMessage, jsonData)
 		WsMutex.Unlock()
 	} else if localdata.UseWS == true && localdata.NodeType == 2 {
-		ws := localdata.WsClients[req.User]
 		WsMutex.Lock()
 		ws.WriteMessage(websocket.TextMessage, jsonData)
 		WsMutex.Unlock()
 	} else {
 		pubsub.Publish(string(jsonData), req.User)
 	}
 }
-func SendSynced(req Request) {
+func SendSynced(req Request, ws *websocket.Conn) {
 	data := map[string]string{
 		"type": "Synced",
 		"user": localdata.GetNodeName(),
@@ -48,7 +46,6 @@ func SendSynced(req Request) {
 	if localdata.WsPeers[req.User] == req.User && localdata.NodeType == 1 {
 		fmt.Println("Sending Synced to " + req.User)
 		WsMutex.Lock()
-		ws := localdata.WsClients[req.User]
 		ws.WriteMessage(websocket.TextMessage, jsonData)
 		WsMutex.Unlock()
 	} else if localdata.UseWS == true && localdata.NodeType == 2 {
@@ -73,15 +70,15 @@ func ReceiveSynced(req Request) {
 	localdata.Lock.Unlock()
 	fmt.Println("Synced with " + req.User)
 }
-func SyncNode(req Request) {
+func SyncNode(req Request, ws *websocket.Conn) {
 	fmt.Println("Syncing with " + req.User)
 	peerName := req.User
 	localdata.Lock.Lock()
 	Nodes[req.User] = true
 	localdata.PeerNames = localdata.RemoveDuplicates(append(localdata.PeerNames, peerName))
 	localdata.NodesStatus[req.User] = "Synced"
 	localdata.Lock.Unlock()
-	SendSynced(req)
+	SendSynced(req, ws)
 	fmt.Println("Synced with " + req.User)
 	Nodes[req.User] = false
 }