Merge pull request #5 from AJackTi/feature/communication_rpc

Feat: Communicating between services using Remote Procedure Calls (RPC)

Author: AJackTi

broker-service/cmd/api/handlers.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"log"
 	"net/http"
+	"net/rpc"
 )
 
 type RequestPayload struct {
@@ -61,6 +62,9 @@ func (app *Config) HandleSubmission(w http.ResponseWriter, r *http.Request) {
 	case "mail":
 		app.sendMail(w, requestPayload.Mail)
 
+	case "log-via-rpc":
+		app.logItemViaRPC(w, requestPayload.Log)
+
 	case "log-event":
 		app.logEventViaRabbit(w, requestPayload.Log)
 
@@ -227,3 +231,35 @@ func (app *Config) pushToQueue(name, msg string) error {
 
 	return nil
 }
+
+type RPCPayload struct {
+	Name string
+	Data string
+}
+
+func (app *Config) logItemViaRPC(w http.ResponseWriter, l LogPayload) {
+	client, err := rpc.Dial("tcp", "logger-service-app:5001")
+	if err != nil {
+		app.errorJSON(w, err)
+		return
+	}
+
+	rpcPayload := RPCPayload{
+		Name: l.Name,
+		Data: l.Data,
+	}
+
+	var result string
+	err = client.Call("RPCServer.LogInfo", rpcPayload, &result)
+	if err != nil {
+		app.errorJSON(w, err)
+		return
+	}
+
+	payload := jsonResponse{
+		Error:   false,
+		Message: result,
+	}
+
+	app.writeJSON(w, http.StatusAccepted, payload)
+}

front-end/cmd/web/templates/test.page.gohtml

@@ -10,6 +10,7 @@
             <a href="javascript:void(0);" id="logBtn" class="btn btn-outline-secondary">Test log</a>
             <a href="javascript:void(0);" id="mailBtn" class="btn btn-outline-secondary">Test mail</a>
             <a href="javascript:void(0);" id="listenerBtn" class="btn btn-outline-secondary">Test listener</a>
+            <a href="javascript:void(0);" id="logViaRPCBtn" class="btn btn-outline-secondary">Test log via RPC</a>
 
             <div id="output" class="mt-5" style="outline: 1px solid silver; padding: 2em;">
                 <span class="text-muted">Output shows here...</span>
@@ -38,6 +39,7 @@
     let logBtn = document.getElementById("logBtn");
     let mailBtn = document.getElementById("mailBtn");
     let listenerBtn = document.getElementById("listenerBtn");
+    let logViaRPCBtn = document.getElementById("logViaRPCBtn");
     let output = document.getElementById("output");
     let sent = document.getElementById("payload");
     let received = document.getElementById("received");
@@ -76,6 +78,40 @@
             })
     })
 
+    logViaRPCBtn.addEventListener("click", function() {
+        const payload = {
+            action: "log-via-rpc",
+            log: {
+                name: "event",
+                data: "Some kind of data",
+            }
+        }
+
+        const headers = new Headers();
+        headers.append("Content-Type", "application/json");
+
+        const body = {
+            method: "POST",
+            body: JSON.stringify(payload),
+            headers: headers,
+        }
+
+        fetch("http:\/\/localhost:8080/handle", body)
+            .then((response) => response.json())
+            .then((data) => {
+                sent.innerHTML = JSON.stringify(payload, undefined, 4);
+                received.innerHTML = JSON.stringify(data, undefined, 4);
+                if (data.error) {
+                    output.innerHTML += ` <br><strong> Error: </strong> ${data.message}`;
+                } else {
+                    output.innerHTML += `<br><strong>Response from broker service</strong>: ${data.message}`;
+                }
+            })
+            .catch((error) => {
+                output.innerHTML += "<br><br>Eror: " + error;
+            })
+    })
+
     mailBtn.addEventListener("click", function() {
         const payload = {
             action: "mail",

logger-service/cmd/api/main.go

@@ -5,7 +5,9 @@ import (
 	"fmt"
 	"log"
 	"log-service/data"
+	"net"
 	"net/http"
+	"net/rpc"
 	"time"
 
 	"go.mongodb.org/mongo-driver/mongo"
@@ -48,8 +50,11 @@ func main() {
 		Models: *data.New(client),
 	}
 
+	// Register the RPC Server
+	err = rpc.Register(new(RPCServer))
+	go app.rpcListen()
+
 	// start web server
-	// go app.serve()
 	log.Println("Starting service on port", webPort)
 	srv := &http.Server{
 		Addr:    fmt.Sprintf(":%s", webPort),
@@ -62,17 +67,25 @@ func main() {
 	}
 }
 
-// func (app *Config) serve() {
-// 	srv := &http.Server{
-// 		Addr:    fmt.Sprintf(":%s", webPort),
-// 		Handler: app.routes(),
-// 	}
-
-// 	err := srv.ListenAndServe()
-// 	if err != nil {
-// 		log.Panic(err)
-// 	}
-// }
+func (app *Config) rpcListen() error {
+	log.Println("Starting RPC server on port", rpcPort)
+	listen, err := net.Listen("tcp", fmt.Sprintf("0.0.0.0:%s", rpcPort))
+	if err != nil {
+		log.Println(err)
+		return err
+	}
+	defer listen.Close()
+
+	for {
+		rpcConn, err := listen.Accept()
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+
+		go rpc.ServeConn(rpcConn)
+	}
+}
 
 func connectToMongo() (*mongo.Client, error) {
 	// create connection options

logger-service/cmd/api/rpc.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"context"
+	"log"
+	"log-service/data"
+	"time"
+)
+
+// RPCServer is the type for our RPC Server. Methods that take this as a receiver are avaiable
+// over RPC, as long as they are exported.
+type RPCServer struct {
+}
+
+// RPCPayload is the type for data we receive from RPC
+type RPCPayload struct {
+	Name string
+	Data string
+}
+
+// LogInfo writes our payload to mongo
+func (r *RPCServer) LogInfo(payload RPCPayload, resp *string) error {
+	collection := client.Database("logs").Collection("logs")
+	_, err := collection.InsertOne(context.TODO(), data.LogEntry{
+		Name:      payload.Name,
+		Data:      payload.Data,
+		CreatedAt: time.Now(),
+	})
+	if err != nil {
+		log.Println("error writing to mongo", err)
+		return err
+	}
+
+	*resp = "Processed payload via RPC: " + payload.Name
+
+	return nil
+}