pubsub (#6)

* inmem pubsub * refactor to use pubsub and channels * non-blocking view.stream * send event on websocket with fallback on http * update plugin version * subscription instead of sub * pubsub redis adapter
livefir · Nov 13, 2022 · 77bd3e3 · 77bd3e3
1 parent e216a0c
commit 77bd3e3
Show file tree

Hide file tree

Showing 17 changed files with 836 additions and 505 deletions.
diff --git a/alpinejs-plugin/package-lock.json b/alpinejs-plugin/package-lock.json
diff --git a/alpinejs-plugin/package.json b/alpinejs-plugin/package.json
@@ -1,6 +1,6 @@
 {
   "name": "@adnaanx/fir",
-  "version": "0.0.7",
+  "version": "0.0.8",
   "repository": {
     "type": "git",
     "url": "https://github.com/adnaan/fir"

diff --git a/alpinejs-plugin/src/index.js b/alpinejs-plugin/src/index.js
@@ -22,7 +22,7 @@ const Plugin = (Alpine) => {
     if (window.location.protocol === "https:") {
         connectURL = `wss://${window.location.host}${window.location.pathname}`
     }
-    websocket(connectURL, [], (patchOperation) => operations[patchOperation.op](patchOperation), updateStore);
+    const socket = websocket(connectURL, [], (patchOperation) => operations[patchOperation.op](patchOperation), updateStore);
 
     Alpine.directive('fir-store', (el, { expression }, { evaluate }) => {
         const val = evaluate(expression)
@@ -64,7 +64,6 @@ const Plugin = (Alpine) => {
                     }
                 });
 
-
                 let formMethod = el.getAttribute("method")
                 if (!formMethod) {
                     formMethod = "get"
@@ -173,45 +172,55 @@ const Plugin = (Alpine) => {
             cancelable: true,
         }
 
-
         el.dispatchEvent(new CustomEvent(startEventName, options))
         if (detail.id) {
             el.dispatchEvent(new CustomEvent(`${startEventName}:${detail.id}`, options))
             el.dispatchEvent(new CustomEvent(`${eventName}:${detail.id}`, options))
         }
-
-        fetch(window.location.pathname, {
-            method: 'POST',
-            headers: {
-                'Content-Type': 'application/json',
-                'X-FIR-MODE': 'event'
-            },
-            body: JSON.stringify({
-                id: id,
-                params: params,
-            }),
-        })
-            .then(response => response.json())
-            .then(patchOperations => {
-                patchOperations.forEach(patchOperation => {
-                    operations[patchOperation.op](patchOperation)
-                });
+        const event = {
+            id: id,
+            params: params,
+        }
+        if (socket.emit(event)) {
+            el.dispatchEvent(new CustomEvent(endEventName, options))
+            if (detail.id) {
+                el.dispatchEvent(new CustomEvent(`${endEventName}:${detail.id}`, options))
+                el.dispatchEvent(new CustomEvent(`${eventName}:${detail.id}`, options))
+            }
+        } else {
+            fetch(window.location.pathname, {
+                method: 'POST',
+                headers: {
+                    'Content-Type': 'application/json',
+                    'X-FIR-MODE': 'event'
+                },
+                body: JSON.stringify({
+                    id: id,
+                    params: params,
+                }),
             })
-            .catch((error) => {
-                console.error(`${endEventName} error: ${error}, detail: ${detail}`, error,);
-            }).finally(() => {
-                el.dispatchEvent(new CustomEvent(endEventName, options))
-                if (detail.id) {
-                    el.dispatchEvent(new CustomEvent(`${endEventName}:${detail.id}`, options))
-                    el.dispatchEvent(new CustomEvent(`${eventName}:${detail.id}`, options))
-                }
-            });
+                .then(response => response.json())
+                .then(patchOperations => {
+                    patchOperations.forEach(patchOperation => {
+                        operations[patchOperation.op](patchOperation)
+                    });
+                })
+                .catch((error) => {
+                    console.error(`${endEventName} error: ${error}, detail: ${detail}`, error,);
+                }).finally(() => {
+                    el.dispatchEvent(new CustomEvent(endEventName, options))
+                    if (detail.id) {
+                        el.dispatchEvent(new CustomEvent(`${endEventName}:${detail.id}`, options))
+                        el.dispatchEvent(new CustomEvent(`${eventName}:${detail.id}`, options))
+                    }
+                });
+        }
+
     }
 
     Alpine.plugin(morph)
 }
 
-
 const isObject = (obj) => {
     return Object.prototype.toString.call(obj) === '[object Object]';
 };

diff --git a/alpinejs-plugin/src/websocket.js b/alpinejs-plugin/src/websocket.js
@@ -1,4 +1,4 @@
-const reopenTimeouts = [2000, 5000, 10000, 30000, 60000];
+const reopenTimeouts = [500, 1000, 1500, 2000, 5000, 10000, 30000, 60000];
 
 export default websocket = (
     url,
@@ -59,7 +59,6 @@ export default websocket = (
             // console.log("socket disconnected")
         }
 
-
         socket.onclose = event => reOpenSocket();
         socket.onmessage = event => {
             try {
@@ -87,4 +86,17 @@ export default websocket = (
     }
 
     openSocket().then(() => { }).catch(e => console.error(e));
+
+    return {
+        emit(value) {
+            const send = () => socket.send(JSON.stringify(value));
+            if (socket.readyState !== WebSocket.OPEN) {
+                openSocket().then(() => { }).catch(e => console.error(e));
+                return false
+            } else {
+                send();
+                return true
+            }
+        }
+    }
 }
diff --git a/channel.go b/channel.go
@@ -0,0 +1,25 @@
+package fir
+
+import (
+	"fmt"
+	"log"
+	"net/http"
+	"strings"
+)
+
+func DefaultChannelFunc(r *http.Request, viewID string) *string {
+	if viewID == "" {
+		viewID = "root"
+		if r.URL.Path != "/" {
+			viewID = strings.Replace(r.URL.Path, "/", "_", -1)
+		}
+	}
+
+	userID, ok := r.Context().Value(UserIDKey).(string)
+	if !ok || userID == "" {
+		log.Printf("warning: no user id in request context. user is anonymous\n")
+		userID = "anonymous"
+	}
+	channel := fmt.Sprintf("%s:%s", userID, viewID)
+	return &channel
+}
diff --git a/cli/cmd/new.go b/cli/cmd/new.go
@@ -1,6 +1,5 @@
 /*
 Copyright © 2022 NAME HERE <EMAIL ADDRESS>
-
 */
 package cmd
 
@@ -37,4 +36,7 @@ func init() {
 	// Cobra supports local flags which will only run when this command
 	// is called directly, e.g.:
 	// newCmd.Flags().BoolP("toggle", "t", false, "Help message for toggle")
+
+	// @todo: litestack project -> sqlite, litestream, nats.io, bulmacss stack deployed to fly.io
+	// @todo: hello project -> a simple project with a custom helloview for local development.
 }