realtime persistence. (#1952)

* feat: websocket persistence

* feat: socketio persistence

* feat: sse persistence

* feat: mqtt persistence

* fix: added types

* fix: typescript issues

Co-authored-by: Andrew Bastin <andrewbastin.k@gmail.com>

packages/hoppscotch-app/components/realtime/Mqtt.vue

@@ -161,6 +161,22 @@ import debounce from "lodash/debounce"
 import { logHoppRequestRunToAnalytics } from "~/helpers/fb/analytics"
 import { useSetting } from "~/newstore/settings"
 import useWindowSize from "~/helpers/utils/useWindowSize"
+import {
+  MQTTEndpoint$,
+  setMQTTEndpoint,
+  MQTTConnectingState$,
+  MQTTConnectionState$,
+  setMQTTConnectingState,
+  setMQTTConnectionState,
+  MQTTSubscriptionState$,
+  setMQTTSubscriptionState,
+  MQTTSocket$,
+  setMQTTSocket,
+  MQTTLog$,
+  setMQTTLog,
+  addMQTTLogLine,
+} from "~/newstore/MQTTSession"
+import { useStream } from "~/helpers/utils/composables"
 
 export default defineComponent({
   components: { Splitpanes, Pane },
@@ -170,21 +186,33 @@ export default defineComponent({
       SIDEBAR: useSetting("SIDEBAR"),
       COLUMN_LAYOUT: useSetting("COLUMN_LAYOUT"),
       SIDEBAR_ON_LEFT: useSetting("SIDEBAR_ON_LEFT"),
+      url: useStream(MQTTEndpoint$, "", setMQTTEndpoint),
+      connectionState: useStream(
+        MQTTConnectionState$,
+        false,
+        setMQTTConnectionState
+      ),
+      connectingState: useStream(
+        MQTTConnectingState$,
+        false,
+        setMQTTConnectingState
+      ),
+      subscriptionState: useStream(
+        MQTTSubscriptionState$,
+        false,
+        setMQTTSubscriptionState
+      ),
+      log: useStream(MQTTLog$, null, setMQTTLog),
+      client: useStream(MQTTSocket$, null, setMQTTSocket),
     }
   },
   data() {
     return {
-      url: "wss://test.mosquitto.org:8081",
       isUrlValid: true,
-      client: null,
       pub_topic: "",
       sub_topic: "",
       msg: "",
-      connectionState: false,
-      connectingState: false,
-      log: null,
       manualDisconnect: false,
-      subscriptionState: false,
       username: "",
       password: "",
     }
@@ -261,7 +289,7 @@ export default defineComponent({
     onConnectionFailure() {
       this.connectingState = false
       this.connectionState = false
-      this.log.push({
+      addMQTTLogLine({
         payload: this.$t("error.something_went_wrong"),
         source: "info",
         color: "#ff5555",
@@ -271,7 +299,7 @@ export default defineComponent({
     onConnectionSuccess() {
       this.connectingState = false
       this.connectionState = true
-      this.log.push({
+      addMQTTLogLine({
         payload: this.$t("state.connected_to", { name: this.url }),
         source: "info",
         color: "var(--accent-color)",
@@ -280,7 +308,7 @@ export default defineComponent({
       this.$toast.success(this.$t("state.connected"))
     },
     onMessageArrived({ payloadString, destinationName }) {
-      this.log.push({
+      addMQTTLogLine({
         payload: `Message: ${payloadString} arrived on topic: ${destinationName}`,
         source: "info",
         color: "var(--accent-color)",
@@ -297,7 +325,7 @@ export default defineComponent({
     disconnect() {
       this.manualDisconnect = true
       this.client.disconnect()
-      this.log.push({
+      addMQTTLogLine({
         payload: this.$t("state.disconnected_from", { name: this.url }),
         source: "info",
         color: "#ff5555",
@@ -318,14 +346,14 @@ export default defineComponent({
     publish() {
       try {
         this.client.publish(this.pub_topic, this.msg, 0, false)
-        this.log.push({
+        addMQTTLogLine({
           payload: `Published message: ${this.msg} to topic: ${this.pub_topic}`,
           ts: new Date().toLocaleTimeString(),
           source: "info",
           color: "var(--accent-color)",
         })
       } catch (e) {
-        this.log.push({
+        addMQTTLogLine({
           payload:
             this.$t("error.something_went_wrong") +
             `while publishing msg: ${this.msg} to topic:  ${this.pub_topic}`,
@@ -349,7 +377,7 @@ export default defineComponent({
           onFailure: this.usubFailure,
         })
       } catch (e) {
-        this.log.push({
+        addMQTTLogLine({
           payload:
             this.$t("error.something_went_wrong") +
             `while subscribing to topic:  ${this.sub_topic}`,
@@ -361,7 +389,7 @@ export default defineComponent({
     },
     usubSuccess() {
       this.subscriptionState = !this.subscriptionState
-      this.log.push({
+      addMQTTLogLine({
         payload:
           `Successfully ` +
           (this.subscriptionState ? "subscribed" : "unsubscribed") +
@@ -372,7 +400,7 @@ export default defineComponent({
       })
     },
     usubFailure() {
-      this.log.push({
+      addMQTTLogLine({
         payload:
           `Failed to ` +
           (this.subscriptionState ? "unsubscribe" : "subscribe") +

packages/hoppscotch-app/components/realtime/Socketio.vue

@@ -89,7 +89,7 @@
           class="hide-scrollbar !overflow-auto"
         >
           <AppSection label="response">
-            <RealtimeLog :title="$t('socketio.log')" :log="communication.log" />
+            <RealtimeLog :title="$t('socketio.log')" :log="log" />
           </AppSection>
         </Pane>
       </Splitpanes>
@@ -188,6 +188,24 @@ import debounce from "lodash/debounce"
 import { logHoppRequestRunToAnalytics } from "~/helpers/fb/analytics"
 import { useSetting } from "~/newstore/settings"
 import useWindowSize from "~/helpers/utils/useWindowSize"
+import {
+  SIOEndpoint$,
+  setSIOEndpoint,
+  SIOVersion$,
+  setSIOVersion,
+  SIOPath$,
+  setSIOPath,
+  SIOConnectionState$,
+  SIOConnectingState$,
+  setSIOConnectionState,
+  setSIOConnectingState,
+  SIOSocket$,
+  setSIOSocket,
+  SIOLog$,
+  setSIOLog,
+  addSIOLogLine,
+} from "~/newstore/SocketIOSession"
+import { useStream } from "~/helpers/utils/composables"
 
 const socketIoClients = {
   v4: ClientV4,
@@ -204,20 +222,27 @@ export default defineComponent({
       COLUMN_LAYOUT: useSetting("COLUMN_LAYOUT"),
       SIDEBAR_ON_LEFT: useSetting("SIDEBAR_ON_LEFT"),
       socketIoClients,
+      url: useStream(SIOEndpoint$, "", setSIOEndpoint),
+      clientVersion: useStream(SIOVersion$, "", setSIOVersion),
+      path: useStream(SIOPath$, "", setSIOPath),
+      connectingState: useStream(
+        SIOConnectingState$,
+        false,
+        setSIOConnectingState
+      ),
+      connectionState: useStream(
+        SIOConnectionState$,
+        false,
+        setSIOConnectionState
+      ),
+      io: useStream(SIOSocket$, null, setSIOSocket),
+      log: useStream(SIOLog$, [], setSIOLog),
     }
   },
   data() {
     return {
-      // default version is set to v4
-      clientVersion: "v4",
-      url: "wss://hoppscotch-socketio.herokuapp.com",
-      path: "/socket.io",
       isUrlValid: true,
-      connectingState: false,
-      connectionState: false,
-      io: null,
       communication: {
-        log: null,
         eventName: "",
         inputs: [""],
       },
@@ -267,7 +292,7 @@ export default defineComponent({
     },
     connect() {
       this.connectingState = true
-      this.communication.log = [
+      this.log = [
         {
           payload: this.$t("state.connecting_to", { name: this.url }),
           source: "info",
@@ -286,7 +311,7 @@ export default defineComponent({
         this.io.on("connect", () => {
           this.connectingState = false
           this.connectionState = true
-          this.communication.log = [
+          this.log = [
             {
               payload: this.$t("state.connected_to", { name: this.url }),
               source: "info",
@@ -298,7 +323,7 @@ export default defineComponent({
         })
         this.io.on("*", ({ data }) => {
           const [eventName, message] = data
-          this.communication.log.push({
+          addSIOLogLine({
             payload: `[${eventName}] ${message ? JSON.stringify(message) : ""}`,
             source: "server",
             ts: new Date().toLocaleTimeString(),
@@ -316,7 +341,7 @@ export default defineComponent({
         this.io.on("disconnect", () => {
           this.connectingState = false
           this.connectionState = false
-          this.communication.log.push({
+          addSIOLogLine({
             payload: this.$t("state.disconnected_from", { name: this.url }),
             source: "info",
             color: "#ff5555",
@@ -340,14 +365,14 @@ export default defineComponent({
       this.disconnect()
       this.connectingState = false
       this.connectionState = false
-      this.communication.log.push({
+      addSIOLogLine({
         payload: this.$t("error.something_went_wrong"),
         source: "info",
         color: "#ff5555",
         ts: new Date().toLocaleTimeString(),
       })
       if (error !== null)
-        this.communication.log.push({
+        addSIOLogLine({
           payload: error,
           source: "info",
           color: "#ff5555",
@@ -369,14 +394,14 @@ export default defineComponent({
       if (this.io) {
         this.io.emit(eventName, ...messages, (data) => {
           // receive response from server
-          this.communication.log.push({
+          addSIOLogLine({
             payload: `[${eventName}] ${JSON.stringify(data)}`,
             source: "server",
             ts: new Date().toLocaleTimeString(),
           })
         })
 
-        this.communication.log.push({
+        addSIOLogLine({
           payload: `[${eventName}] ${JSON.stringify(messages)}`,
           source: "client",
           ts: new Date().toLocaleTimeString(),

packages/hoppscotch-app/components/realtime/Sse.vue

@@ -48,7 +48,7 @@
       <AppSection label="response">
         <ul>
           <li>
-            <RealtimeLog :title="$t('sse.log')" :log="events.log" />
+            <RealtimeLog :title="$t('sse.log')" :log="log" />
             <div id="result"></div>
           </li>
         </ul>
@@ -64,26 +64,47 @@ import "splitpanes/dist/splitpanes.css"
 import debounce from "lodash/debounce"
 import { logHoppRequestRunToAnalytics } from "~/helpers/fb/analytics"
 import { useSetting } from "~/newstore/settings"
+import {
+  SSEEndpoint$,
+  setSSEEndpoint,
+  SSEEventType$,
+  setSSEEventType,
+  SSESocket$,
+  setSSESocket,
+  SSEConnectingState$,
+  SSEConnectionState$,
+  setSSEConnectionState,
+  setSSEConnectingState,
+  SSELog$,
+  setSSELog,
+  addSSELogLine,
+} from "~/newstore/SSESession"
+import { useStream } from "~/helpers/utils/composables"
 
 export default defineComponent({
   components: { Splitpanes, Pane },
   setup() {
     return {
       COLUMN_LAYOUT: useSetting("COLUMN_LAYOUT"),
+      connectionSSEState: useStream(
+        SSEConnectionState$,
+        false,
+        setSSEConnectionState
+      ),
+      connectingState: useStream(
+        SSEConnectingState$,
+        false,
+        setSSEConnectingState
+      ),
+      server: useStream(SSEEndpoint$, "", setSSEEndpoint),
+      eventType: useStream(SSEEventType$, "", setSSEEventType),
+      sse: useStream(SSESocket$, null, setSSESocket),
+      log: useStream(SSELog$, [], setSSELog),
     }
   },
   data() {
     return {
-      connectionSSEState: false,
-      connectingState: false,
-      server: "https://express-eventsource.herokuapp.com/events",
       isUrlValid: true,
-      sse: null,
-      events: {
-        log: null,
-        input: "",
-      },
-      eventType: "data",
     }
   },
   computed: {
@@ -120,7 +141,7 @@ export default defineComponent({
     },
     start() {
       this.connectingState = true
-      this.events.log = [
+      this.log = [
         {
           payload: this.$t("state.connecting_to", { name: this.server }),
           source: "info",
@@ -133,7 +154,7 @@ export default defineComponent({
           this.sse.onopen = () => {
             this.connectingState = false
             this.connectionSSEState = true
-            this.events.log = [
+            this.log = [
               {
                 payload: this.$t("state.connected_to", { name: this.server }),
                 source: "info",
@@ -148,7 +169,7 @@ export default defineComponent({
           }
           this.sse.onclose = () => {
             this.connectionSSEState = false
-            this.events.log.push({
+            addSSELogLine({
               payload: this.$t("state.disconnected_from", {
                 name: this.server,
               }),
@@ -159,7 +180,7 @@ export default defineComponent({
             this.$toast.error(this.$t("state.disconnected"))
           }
           this.sse.addEventListener(this.eventType, ({ data }) => {
-            this.events.log.push({
+            addSSELogLine({
               payload: data,
               source: "server",
               ts: new Date().toLocaleTimeString(),
@@ -170,7 +191,7 @@ export default defineComponent({
           this.$toast.error(this.$t("error.something_went_wrong"))
         }
       } else {
-        this.events.log = [
+        this.log = [
           {
             payload: this.$t("error.browser_support_sse"),
             source: "info",
@@ -187,14 +208,14 @@ export default defineComponent({
     handleSSEError(error) {
       this.stop()
       this.connectionSSEState = false
-      this.events.log.push({
+      addSSELogLine({
         payload: this.$t("error.something_went_wrong"),
         source: "info",
         color: "#ff5555",
         ts: new Date().toLocaleTimeString(),
       })
       if (error !== null)
-        this.events.log.push({
+        addSSELogLine({
           payload: error,
           source: "info",
           color: "#ff5555",

packages/hoppscotch-app/components/realtime/Websocket.vue

@@ -76,6 +76,12 @@
                 name="message"
                 type="text"
                 autocomplete="off"
+                @change="
+                  updateProtocol(index, {
+                    value: $event.target.value,
+                    active: protocol.active,
+                  })
+                "
               />
               <span>
                 <ButtonSecondary
@@ -96,9 +102,10 @@
                   "
                   color="green"
                   @click.native="
-                    protocol.active = protocol.hasOwnProperty('active')
+                    updateProtocol(index, {
-                      ? !protocol.active
+                      value: protocol.value,
-                      : false
+                      active: !protocol.active,
+                    })
                   "
                 />
               </span>
@@ -133,10 +140,7 @@
           class="hide-scrollbar !overflow-auto"
         >
           <AppSection label="response">
-            <RealtimeLog
+            <RealtimeLog :title="$t('websocket.log')" :log="log" />
-              :title="$t('websocket.log')"
-              :log="communication.log"
-            />
           </AppSection>
         </Pane>
       </Splitpanes>
@@ -191,6 +195,26 @@ import debounce from "lodash/debounce"
 import { logHoppRequestRunToAnalytics } from "~/helpers/fb/analytics"
 import useWindowSize from "~/helpers/utils/useWindowSize"
 import { useSetting } from "~/newstore/settings"
+import {
+  setWSEndpoint,
+  WSEndpoint$,
+  WSProtocols$,
+  setWSProtocols,
+  addWSProtocol,
+  deleteWSProtocol,
+  updateWSProtocol,
+  deleteAllWSProtocols,
+  WSSocket$,
+  setWSSocket,
+  setWSConnectionState,
+  setWSConnectingState,
+  WSConnectionState$,
+  WSConnectingState$,
+  addWSLogLine,
+  WSLog$,
+  setWSLog,
+} from "~/newstore/WebSocketSession"
+import { useStream } from "~/helpers/utils/composables"
 
 export default defineComponent({
   components: { Splitpanes, Pane },
@@ -200,21 +224,29 @@ export default defineComponent({
       SIDEBAR: useSetting("SIDEBAR"),
       COLUMN_LAYOUT: useSetting("COLUMN_LAYOUT"),
       SIDEBAR_ON_LEFT: useSetting("SIDEBAR_ON_LEFT"),
+      url: useStream(WSEndpoint$, "", setWSEndpoint),
+      protocols: useStream(WSProtocols$, [], setWSProtocols),
+      connectionState: useStream(
+        WSConnectionState$,
+        false,
+        setWSConnectionState
+      ),
+      connectingState: useStream(
+        WSConnectingState$,
+        false,
+        setWSConnectingState
+      ),
+      socket: useStream(WSSocket$, null, setWSSocket),
+      log: useStream(WSLog$, [], setWSLog),
     }
   },
   data() {
     return {
-      connectionState: false,
-      connectingState: false,
-      url: "wss://hoppscotch-websocket.herokuapp.com",
       isUrlValid: true,
-      socket: null,
       communication: {
-        log: null,
         input: "",
       },
       currentIndex: -1, // index of the message log array to put in input box
-      protocols: [],
       activeProtocols: [],
     }
   },
@@ -251,7 +283,7 @@ export default defineComponent({
   },
   methods: {
     clearContent() {
-      this.protocols = []
+      deleteAllWSProtocols()
     },
     debouncer: debounce(function () {
       this.worker.postMessage({ type: "ws", url: this.url })
@@ -266,7 +298,7 @@ export default defineComponent({
       else return this.disconnect()
     },
     connect() {
-      this.communication.log = [
+      this.log = [
         {
           payload: this.$t("state.connecting_to", { name: this.url }),
           source: "info",
@@ -279,7 +311,7 @@ export default defineComponent({
         this.socket.onopen = () => {
           this.connectingState = false
           this.connectionState = true
-          this.communication.log = [
+          this.log = [
             {
               payload: this.$t("state.connected_to", { name: this.url }),
               source: "info",
@@ -294,7 +326,7 @@ export default defineComponent({
         }
         this.socket.onclose = () => {
           this.connectionState = false
-          this.communication.log.push({
+          addWSLogLine({
             payload: this.$t("state.disconnected_from", { name: this.url }),
             source: "info",
             color: "#ff5555",
@@ -303,7 +335,7 @@ export default defineComponent({
           this.$toast.error(this.$t("state.disconnected"))
         }
         this.socket.onmessage = ({ data }) => {
-          this.communication.log.push({
+          addWSLogLine({
             payload: data,
             source: "server",
             ts: new Date().toLocaleTimeString(),
@@ -328,14 +360,14 @@ export default defineComponent({
     handleError(error) {
       this.disconnect()
       this.connectionState = false
-      this.communication.log.push({
+      addWSLogLine({
         payload: this.$t("error.something_went_wrong"),
         source: "info",
         color: "#ff5555",
         ts: new Date().toLocaleTimeString(),
       })
       if (error !== null)
-        this.communication.log.push({
+        addWSLogLine({
           payload: error,
           source: "info",
           color: "#ff5555",
@@ -345,7 +377,7 @@ export default defineComponent({
     sendMessage() {
       const message = this.communication.input
       this.socket.send(message)
-      this.communication.log.push({
+      addWSLogLine({
         payload: message,
         source: "client",
         ts: new Date().toLocaleTimeString(),
@@ -353,7 +385,7 @@ export default defineComponent({
       this.communication.input = ""
     },
     walkHistory(direction) {
-      const clientMessages = this.communication.log.filter(
+      const clientMessages = this.log.filter(
         ({ source }) => source === "client"
       )
       const length = clientMessages.length
@@ -389,11 +421,11 @@ export default defineComponent({
       }
     },
     addProtocol() {
-      this.protocols.push({ value: "", active: true })
+      addWSProtocol({ value: "", active: true })
     },
     deleteProtocol({ index }) {
       const oldProtocols = this.protocols.slice()
-      this.$delete(this.protocols, index)
+      deleteWSProtocol(index)
       this.$toast.success(this.$t("state.deleted"), {
         action: {
           text: this.$t("action.undo"),
@@ -405,6 +437,9 @@ export default defineComponent({
         },
       })
     },
+    updateProtocol(index, updated) {
+      updateWSProtocol(index, updated)
+    },
   },
 })
 </script>

packages/hoppscotch-app/helpers/types/HoppRealtimeLog.ts

@@ -0,0 +1,8 @@
+export type HoppRealtimeLogLine = {
+  payload: string
+  source: string
+  color?: string
+  ts: string
+}
+
+export type HoppRealtimeLog = HoppRealtimeLogLine[]

packages/hoppscotch-app/newstore/MQTTSession.ts

@@ -0,0 +1,190 @@
+import { pluck, distinctUntilChanged } from "rxjs/operators"
+import { Client as MQTTClient } from "paho-mqtt"
+import DispatchingStore, { defineDispatchers } from "./DispatchingStore"
+import {
+  HoppRealtimeLog,
+  HoppRealtimeLogLine,
+} from "~/helpers/types/HoppRealtimeLog"
+
+type HoppMQTTRequest = {
+  endpoint: string
+}
+
+type HoppMQTTSession = {
+  request: HoppMQTTRequest
+  connectingState: boolean
+  connectionState: boolean
+  subscriptionState: boolean
+  log: HoppRealtimeLog
+  socket: MQTTClient | null
+}
+
+const defaultMQTTRequest: HoppMQTTRequest = {
+  endpoint: "wss://test.mosquitto.org:8081",
+}
+
+const defaultMQTTSession: HoppMQTTSession = {
+  request: defaultMQTTRequest,
+  connectionState: false,
+  connectingState: false,
+  subscriptionState: false,
+  socket: null,
+  log: [],
+}
+
+const dispatchers = defineDispatchers({
+  setRequest(
+    _: HoppMQTTSession,
+    { newRequest }: { newRequest: HoppMQTTRequest }
+  ) {
+    return {
+      request: newRequest,
+    }
+  },
+  setEndpoint(_: HoppMQTTSession, { newEndpoint }: { newEndpoint: string }) {
+    return {
+      request: {
+        endpoint: newEndpoint,
+      },
+    }
+  },
+  setSocket(_: HoppMQTTSession, { socket }: { socket: MQTTClient }) {
+    return {
+      socket,
+    }
+  },
+  setConnectionState(_: HoppMQTTSession, { state }: { state: boolean }) {
+    return {
+      connectionState: state,
+    }
+  },
+  setConnectingState(_: HoppMQTTSession, { state }: { state: boolean }) {
+    return {
+      connectingState: state,
+    }
+  },
+  setSubscriptionState(_: HoppMQTTSession, { state }: { state: boolean }) {
+    return {
+      subscriptionState: state,
+    }
+  },
+  setLog(_: HoppMQTTSession, { log }: { log: HoppRealtimeLog }) {
+    return {
+      log,
+    }
+  },
+  addLogLine(curr: HoppMQTTSession, { line }: { line: HoppRealtimeLogLine }) {
+    return {
+      log: [...curr.log, line],
+    }
+  },
+})
+
+const MQTTSessionStore = new DispatchingStore(defaultMQTTSession, dispatchers)
+
+export function setMQTTRequest(newRequest?: HoppMQTTRequest) {
+  MQTTSessionStore.dispatch({
+    dispatcher: "setRequest",
+    payload: {
+      newRequest: newRequest ?? defaultMQTTRequest,
+    },
+  })
+}
+
+export function setMQTTEndpoint(newEndpoint: string) {
+  MQTTSessionStore.dispatch({
+    dispatcher: "setEndpoint",
+    payload: {
+      newEndpoint,
+    },
+  })
+}
+
+export function setMQTTSocket(socket: MQTTClient) {
+  MQTTSessionStore.dispatch({
+    dispatcher: "setSocket",
+    payload: {
+      socket,
+    },
+  })
+}
+
+export function setMQTTConnectionState(state: boolean) {
+  MQTTSessionStore.dispatch({
+    dispatcher: "setConnectionState",
+    payload: {
+      state,
+    },
+  })
+}
+
+export function setMQTTConnectingState(state: boolean) {
+  MQTTSessionStore.dispatch({
+    dispatcher: "setConnectingState",
+    payload: {
+      state,
+    },
+  })
+}
+
+export function setMQTTSubscriptionState(state: boolean) {
+  MQTTSessionStore.dispatch({
+    dispatcher: "setSubscriptionState",
+    payload: {
+      state,
+    },
+  })
+}
+
+export function setMQTTLog(log: HoppRealtimeLog) {
+  MQTTSessionStore.dispatch({
+    dispatcher: "setLog",
+    payload: {
+      log,
+    },
+  })
+}
+
+export function addMQTTLogLine(line: HoppRealtimeLogLine) {
+  MQTTSessionStore.dispatch({
+    dispatcher: "addLogLine",
+    payload: {
+      line,
+    },
+  })
+}
+
+export const MQTTRequest$ = MQTTSessionStore.subject$.pipe(
+  pluck("request"),
+  distinctUntilChanged()
+)
+
+export const MQTTEndpoint$ = MQTTSessionStore.subject$.pipe(
+  pluck("request", "endpoint"),
+  distinctUntilChanged()
+)
+
+export const MQTTConnectingState$ = MQTTSessionStore.subject$.pipe(
+  pluck("connectingState"),
+  distinctUntilChanged()
+)
+
+export const MQTTConnectionState$ = MQTTSessionStore.subject$.pipe(
+  pluck("connectionState"),
+  distinctUntilChanged()
+)
+
+export const MQTTSubscriptionState$ = MQTTSessionStore.subject$.pipe(
+  pluck("subscriptionState"),
+  distinctUntilChanged()
+)
+
+export const MQTTSocket$ = MQTTSessionStore.subject$.pipe(
+  pluck("socket"),
+  distinctUntilChanged()
+)
+
+export const MQTTLog$ = MQTTSessionStore.subject$.pipe(
+  pluck("log"),
+  distinctUntilChanged()
+)

packages/hoppscotch-app/newstore/SSESession.ts

@@ -0,0 +1,192 @@
+import { pluck, distinctUntilChanged } from "rxjs/operators"
+import DispatchingStore, { defineDispatchers } from "./DispatchingStore"
+import {
+  HoppRealtimeLog,
+  HoppRealtimeLogLine,
+} from "~/helpers/types/HoppRealtimeLog"
+
+type HoppSSERequest = {
+  endpoint: string
+  eventType: string
+}
+
+type HoppSSESession = {
+  request: HoppSSERequest
+  connectingState: boolean
+  connectionState: boolean
+  log: HoppRealtimeLog
+  socket: EventSource | null
+}
+
+const defaultSSERequest: HoppSSERequest = {
+  endpoint: "https://express-eventsource.herokuapp.com/events",
+  eventType: "data",
+}
+
+const defaultSSESession: HoppSSESession = {
+  request: defaultSSERequest,
+  connectionState: false,
+  connectingState: false,
+  socket: null,
+  log: [],
+}
+
+const dispatchers = defineDispatchers({
+  setRequest(
+    _: HoppSSESession,
+    { newRequest }: { newRequest: HoppSSERequest }
+  ) {
+    return {
+      request: newRequest,
+    }
+  },
+  setEndpoint(curr: HoppSSESession, { newEndpoint }: { newEndpoint: string }) {
+    return {
+      request: {
+        eventType: curr.request.eventType,
+        endpoint: newEndpoint,
+      },
+    }
+  },
+  setEventType(curr: HoppSSESession, { newType }: { newType: string }) {
+    return {
+      request: {
+        endpoint: curr.request.endpoint,
+        eventType: newType,
+      },
+    }
+  },
+  setSocket(_: HoppSSESession, { socket }: { socket: EventSource }) {
+    return {
+      socket,
+    }
+  },
+  setConnectionState(_: HoppSSESession, { state }: { state: boolean }) {
+    return {
+      connectionState: state,
+    }
+  },
+  setConnectingState(_: HoppSSESession, { state }: { state: boolean }) {
+    return {
+      connectingState: state,
+    }
+  },
+  setLog(_: HoppSSESession, { log }: { log: HoppRealtimeLog }) {
+    return {
+      log,
+    }
+  },
+  addLogLine(curr: HoppSSESession, { line }: { line: HoppRealtimeLogLine }) {
+    return {
+      log: [...curr.log, line],
+    }
+  },
+})
+
+const SSESessionStore = new DispatchingStore(defaultSSESession, dispatchers)
+
+export function setSSERequest(newRequest?: HoppSSERequest) {
+  SSESessionStore.dispatch({
+    dispatcher: "setRequest",
+    payload: {
+      newRequest: newRequest ?? defaultSSERequest,
+    },
+  })
+}
+
+export function setSSEEndpoint(newEndpoint: string) {
+  SSESessionStore.dispatch({
+    dispatcher: "setEndpoint",
+    payload: {
+      newEndpoint,
+    },
+  })
+}
+
+export function setSSEEventType(newType: string) {
+  SSESessionStore.dispatch({
+    dispatcher: "setEventType",
+    payload: {
+      newType,
+    },
+  })
+}
+
+export function setSSESocket(socket: EventSource) {
+  SSESessionStore.dispatch({
+    dispatcher: "setSocket",
+    payload: {
+      socket,
+    },
+  })
+}
+
+export function setSSEConnectionState(state: boolean) {
+  SSESessionStore.dispatch({
+    dispatcher: "setConnectionState",
+    payload: {
+      state,
+    },
+  })
+}
+export function setSSEConnectingState(state: boolean) {
+  SSESessionStore.dispatch({
+    dispatcher: "setConnectingState",
+    payload: {
+      state,
+    },
+  })
+}
+
+export function setSSELog(log: HoppRealtimeLog) {
+  SSESessionStore.dispatch({
+    dispatcher: "setLog",
+    payload: {
+      log,
+    },
+  })
+}
+
+export function addSSELogLine(line: HoppRealtimeLogLine) {
+  SSESessionStore.dispatch({
+    dispatcher: "addLogLine",
+    payload: {
+      line,
+    },
+  })
+}
+
+export const SSERequest$ = SSESessionStore.subject$.pipe(
+  pluck("request"),
+  distinctUntilChanged()
+)
+
+export const SSEEndpoint$ = SSESessionStore.subject$.pipe(
+  pluck("request", "endpoint"),
+  distinctUntilChanged()
+)
+
+export const SSEEventType$ = SSESessionStore.subject$.pipe(
+  pluck("request", "eventType"),
+  distinctUntilChanged()
+)
+
+export const SSEConnectingState$ = SSESessionStore.subject$.pipe(
+  pluck("connectingState"),
+  distinctUntilChanged()
+)
+
+export const SSEConnectionState$ = SSESessionStore.subject$.pipe(
+  pluck("connectionState"),
+  distinctUntilChanged()
+)
+
+export const SSESocket$ = SSESessionStore.subject$.pipe(
+  pluck("socket"),
+  distinctUntilChanged()
+)
+
+export const SSELog$ = SSESessionStore.subject$.pipe(
+  pluck("log"),
+  distinctUntilChanged()
+)

packages/hoppscotch-app/newstore/SocketIOSession.ts

@@ -0,0 +1,221 @@
+import { pluck, distinctUntilChanged } from "rxjs/operators"
+import { Socket as SocketV2 } from "socket.io-client-v2"
+import { Socket as SocketV3 } from "socket.io-client-v3"
+import { Socket as SocketV4 } from "socket.io-client-v4"
+import DispatchingStore, { defineDispatchers } from "./DispatchingStore"
+import {
+  HoppRealtimeLog,
+  HoppRealtimeLogLine,
+} from "~/helpers/types/HoppRealtimeLog"
+
+type SocketIO = SocketV2 | SocketV3 | SocketV4
+
+type HoppSIORequest = {
+  endpoint: string
+  path: string
+  version: string
+}
+
+type HoppSIOSession = {
+  request: HoppSIORequest
+  connectingState: boolean
+  connectionState: boolean
+  log: HoppRealtimeLog
+  socket: SocketIO | null
+}
+
+const defaultSIORequest: HoppSIORequest = {
+  endpoint: "wss://hoppscotch-socketio.herokuapp.com",
+  path: "/socket.io",
+  version: "v4",
+}
+
+const defaultSIOSession: HoppSIOSession = {
+  request: defaultSIORequest,
+  connectionState: false,
+  connectingState: false,
+  socket: null,
+  log: [],
+}
+
+const dispatchers = defineDispatchers({
+  setRequest(
+    _: HoppSIOSession,
+    { newRequest }: { newRequest: HoppSIORequest }
+  ) {
+    return {
+      request: newRequest,
+    }
+  },
+  setEndpoint(curr: HoppSIOSession, { newEndpoint }: { newEndpoint: string }) {
+    return {
+      request: {
+        ...curr.request,
+        endpoint: newEndpoint,
+      },
+    }
+  },
+  setPath(curr: HoppSIOSession, { newPath }: { newPath: string }) {
+    return {
+      request: {
+        ...curr.request,
+        path: newPath,
+      },
+    }
+  },
+  setVersion(curr: HoppSIOSession, { newVersion }: { newVersion: string }) {
+    return {
+      request: {
+        ...curr.request,
+        version: newVersion,
+      },
+    }
+  },
+  setSocket(_: HoppSIOSession, { socket }: { socket: SocketIO }) {
+    return {
+      socket,
+    }
+  },
+  setConnectionState(_: HoppSIOSession, { state }: { state: boolean }) {
+    return {
+      connectionState: state,
+    }
+  },
+  setConnectingState(_: HoppSIOSession, { state }: { state: boolean }) {
+    return {
+      connectingState: state,
+    }
+  },
+  setLog(_: HoppSIOSession, { log }: { log: HoppRealtimeLog }) {
+    return {
+      log,
+    }
+  },
+  addLogLine(curr: HoppSIOSession, { line }: { line: HoppRealtimeLogLine }) {
+    return {
+      log: [...curr.log, line],
+    }
+  },
+})
+
+const SIOSessionStore = new DispatchingStore(defaultSIOSession, dispatchers)
+
+export function setSIORequest(newRequest?: HoppSIORequest) {
+  SIOSessionStore.dispatch({
+    dispatcher: "setRequest",
+    payload: {
+      newRequest: newRequest ?? defaultSIORequest,
+    },
+  })
+}
+
+export function setSIOEndpoint(newEndpoint: string) {
+  SIOSessionStore.dispatch({
+    dispatcher: "setEndpoint",
+    payload: {
+      newEndpoint,
+    },
+  })
+}
+
+export function setSIOVersion(newVersion: string) {
+  SIOSessionStore.dispatch({
+    dispatcher: "setVersion",
+    payload: {
+      newVersion,
+    },
+  })
+}
+
+export function setSIOPath(newPath: string) {
+  SIOSessionStore.dispatch({
+    dispatcher: "setPath",
+    payload: {
+      newPath,
+    },
+  })
+}
+
+export function setSIOSocket(socket: SocketIO) {
+  SIOSessionStore.dispatch({
+    dispatcher: "setSocket",
+    payload: {
+      socket,
+    },
+  })
+}
+
+export function setSIOConnectionState(state: boolean) {
+  SIOSessionStore.dispatch({
+    dispatcher: "setConnectionState",
+    payload: {
+      state,
+    },
+  })
+}
+export function setSIOConnectingState(state: boolean) {
+  SIOSessionStore.dispatch({
+    dispatcher: "setConnectingState",
+    payload: {
+      state,
+    },
+  })
+}
+
+export function setSIOLog(log: HoppRealtimeLog) {
+  SIOSessionStore.dispatch({
+    dispatcher: "setLog",
+    payload: {
+      log,
+    },
+  })
+}
+
+export function addSIOLogLine(line: HoppRealtimeLogLine) {
+  SIOSessionStore.dispatch({
+    dispatcher: "addLogLine",
+    payload: {
+      line,
+    },
+  })
+}
+
+export const SIORequest$ = SIOSessionStore.subject$.pipe(
+  pluck("request"),
+  distinctUntilChanged()
+)
+
+export const SIOEndpoint$ = SIOSessionStore.subject$.pipe(
+  pluck("request", "endpoint"),
+  distinctUntilChanged()
+)
+
+export const SIOVersion$ = SIOSessionStore.subject$.pipe(
+  pluck("request", "version"),
+  distinctUntilChanged()
+)
+
+export const SIOPath$ = SIOSessionStore.subject$.pipe(
+  pluck("request", "path"),
+  distinctUntilChanged()
+)
+
+export const SIOConnectingState$ = SIOSessionStore.subject$.pipe(
+  pluck("connectingState"),
+  distinctUntilChanged()
+)
+
+export const SIOConnectionState$ = SIOSessionStore.subject$.pipe(
+  pluck("connectionState"),
+  distinctUntilChanged()
+)
+
+export const SIOSocket$ = SIOSessionStore.subject$.pipe(
+  pluck("socket"),
+  distinctUntilChanged()
+)
+
+export const SIOLog$ = SIOSessionStore.subject$.pipe(
+  pluck("log"),
+  distinctUntilChanged()
+)

packages/hoppscotch-app/newstore/WebSocketSession.ts

@@ -0,0 +1,275 @@
+import { pluck, distinctUntilChanged } from "rxjs/operators"
+import DispatchingStore, { defineDispatchers } from "./DispatchingStore"
+import {
+  HoppRealtimeLog,
+  HoppRealtimeLogLine,
+} from "~/helpers/types/HoppRealtimeLog"
+
+type HoppWSProtocol = {
+  value: string
+  active: boolean
+}
+
+type HoppWSRequest = {
+  endpoint: string
+  protocols: HoppWSProtocol[]
+}
+
+export type HoppWSSession = {
+  request: HoppWSRequest
+  connectingState: boolean
+  connectionState: boolean
+  log: HoppRealtimeLog
+  socket: WebSocket | null
+}
+
+const defaultWSRequest: HoppWSRequest = {
+  endpoint: "wss://hoppscotch-websocket.herokuapp.com",
+  protocols: [],
+}
+
+const defaultWSSession: HoppWSSession = {
+  request: defaultWSRequest,
+  connectionState: false,
+  connectingState: false,
+  socket: null,
+  log: [],
+}
+
+const dispatchers = defineDispatchers({
+  setRequest(_: HoppWSSession, { newRequest }: { newRequest: HoppWSRequest }) {
+    return {
+      request: newRequest,
+    }
+  },
+  setEndpoint(curr: HoppWSSession, { newEndpoint }: { newEndpoint: string }) {
+    return {
+      request: {
+        protocols: curr.request.protocols,
+        endpoint: newEndpoint,
+      },
+    }
+  },
+  setProtocols(
+    curr: HoppWSSession,
+    { protocols }: { protocols: HoppWSProtocol[] }
+  ) {
+    return {
+      request: {
+        protocols,
+        endpoint: curr.request.endpoint,
+      },
+    }
+  },
+  addProtocol(curr: HoppWSSession, { protocol }: { protocol: HoppWSProtocol }) {
+    return {
+      request: {
+        endpoint: curr.request.endpoint,
+        protocols: [...curr.request.protocols, protocol],
+      },
+    }
+  },
+  deleteProtocol(curr: HoppWSSession, { index }: { index: number }) {
+    return {
+      request: {
+        endpoint: curr.request.endpoint,
+        protocols: curr.request.protocols.filter((_, idx) => index !== idx),
+      },
+    }
+  },
+  deleteAllProtocols(curr: HoppWSSession) {
+    return {
+      request: {
+        endpoint: curr.request.endpoint,
+        protocols: [],
+      },
+    }
+  },
+  updateProtocol(
+    curr: HoppWSSession,
+    {
+      index,
+      updatedProtocol,
+    }: { index: number; updatedProtocol: HoppWSProtocol }
+  ) {
+    return {
+      request: {
+        endpoint: curr.request.endpoint,
+        protocols: curr.request.protocols.map((proto, idx) => {
+          return index === idx ? updatedProtocol : proto
+        }),
+      },
+    }
+  },
+  setSocket(_: HoppWSSession, { socket }: { socket: WebSocket }) {
+    return {
+      socket,
+    }
+  },
+  setConnectionState(_: HoppWSSession, { state }: { state: boolean }) {
+    return {
+      connectionState: state,
+    }
+  },
+  setConnectingState(_: HoppWSSession, { state }: { state: boolean }) {
+    return {
+      connectingState: state,
+    }
+  },
+  setLog(_: HoppWSSession, { log }: { log: HoppRealtimeLog }) {
+    return {
+      log,
+    }
+  },
+  addLogLine(curr: HoppWSSession, { line }: { line: HoppRealtimeLogLine }) {
+    return {
+      log: [...curr.log, line],
+    }
+  },
+})
+
+const WSSessionStore = new DispatchingStore(defaultWSSession, dispatchers)
+
+export function setWSRequest(newRequest?: HoppWSRequest) {
+  WSSessionStore.dispatch({
+    dispatcher: "setRequest",
+    payload: {
+      newRequest: newRequest ?? defaultWSRequest,
+    },
+  })
+}
+
+export function setWSEndpoint(newEndpoint: string) {
+  WSSessionStore.dispatch({
+    dispatcher: "setEndpoint",
+    payload: {
+      newEndpoint,
+    },
+  })
+}
+
+export function setWSProtocols(protocols: HoppWSProtocol[]) {
+  WSSessionStore.dispatch({
+    dispatcher: "setProtocols",
+    payload: {
+      protocols,
+    },
+  })
+}
+
+export function addWSProtocol(protocol: HoppWSProtocol) {
+  WSSessionStore.dispatch({
+    dispatcher: "addProtocol",
+    payload: {
+      protocol,
+    },
+  })
+}
+
+export function deleteWSProtocol(index: number) {
+  WSSessionStore.dispatch({
+    dispatcher: "deleteProtocol",
+    payload: {
+      index,
+    },
+  })
+}
+
+export function deleteAllWSProtocols() {
+  WSSessionStore.dispatch({
+    dispatcher: "deleteAllProtocols",
+    payload: {},
+  })
+}
+
+export function updateWSProtocol(
+  index: number,
+  updatedProtocol: HoppWSProtocol
+) {
+  WSSessionStore.dispatch({
+    dispatcher: "updateProtocol",
+    payload: {
+      index,
+      updatedProtocol,
+    },
+  })
+}
+
+export function setWSSocket(socket: WebSocket) {
+  WSSessionStore.dispatch({
+    dispatcher: "setSocket",
+    payload: {
+      socket,
+    },
+  })
+}
+
+export function setWSConnectionState(state: boolean) {
+  WSSessionStore.dispatch({
+    dispatcher: "setConnectionState",
+    payload: {
+      state,
+    },
+  })
+}
+export function setWSConnectingState(state: boolean) {
+  WSSessionStore.dispatch({
+    dispatcher: "setConnectingState",
+    payload: {
+      state,
+    },
+  })
+}
+
+export function setWSLog(log: HoppRealtimeLog) {
+  WSSessionStore.dispatch({
+    dispatcher: "setLog",
+    payload: {
+      log,
+    },
+  })
+}
+
+export function addWSLogLine(line: HoppRealtimeLogLine) {
+  WSSessionStore.dispatch({
+    dispatcher: "addLogLine",
+    payload: {
+      line,
+    },
+  })
+}
+
+export const WSRequest$ = WSSessionStore.subject$.pipe(
+  pluck("request"),
+  distinctUntilChanged()
+)
+
+export const WSEndpoint$ = WSSessionStore.subject$.pipe(
+  pluck("request", "endpoint"),
+  distinctUntilChanged()
+)
+
+export const WSProtocols$ = WSSessionStore.subject$.pipe(
+  pluck("request", "protocols"),
+  distinctUntilChanged()
+)
+
+export const WSConnectingState$ = WSSessionStore.subject$.pipe(
+  pluck("connectingState"),
+  distinctUntilChanged()
+)
+
+export const WSConnectionState$ = WSSessionStore.subject$.pipe(
+  pluck("connectionState"),
+  distinctUntilChanged()
+)
+
+export const WSSocket$ = WSSessionStore.subject$.pipe(
+  pluck("socket"),
+  distinctUntilChanged()
+)
+
+export const WSLog$ = WSSessionStore.subject$.pipe(
+  pluck("log"),
+  distinctUntilChanged()
+)

packages/hoppscotch-app/newstore/localpersistence.ts

@@ -40,6 +40,10 @@ import {
   setCurrentEnvironment,
 } from "./environments"
 import { restRequest$, setRESTRequest } from "./RESTSession"
+import { WSRequest$, setWSRequest } from "./WebSocketSession"
+import { SIORequest$, setSIORequest } from "./SocketIOSession"
+import { SSERequest$, setSSERequest } from "./SSESession"
+import { MQTTRequest$, setMQTTRequest } from "./MQTTSession"
 import { translateToNewRequest } from "~/helpers/types/HoppRESTRequest"
 
 function checkAndMigrateOldSettings() {
@@ -209,6 +213,54 @@ function setupSelectedEnvPersistence() {
   })
 }
 
+function setupWebsocketPersistence() {
+  const request = JSON.parse(
+    window.localStorage.getItem("WebsocketRequest") || "null"
+  )
+
+  setWSRequest(request)
+
+  WSRequest$.subscribe((req) => {
+    window.localStorage.setItem("WebsocketRequest", JSON.stringify(req))
+  })
+}
+
+function setupSocketIOPersistence() {
+  const request = JSON.parse(
+    window.localStorage.getItem("SocketIORequest") || "null"
+  )
+
+  setSIORequest(request)
+
+  SIORequest$.subscribe((req) => {
+    window.localStorage.setItem("SocketIORequest", JSON.stringify(req))
+  })
+}
+
+function setupSSEPersistence() {
+  const request = JSON.parse(
+    window.localStorage.getItem("SSERequest") || "null"
+  )
+
+  setSSERequest(request)
+
+  SSERequest$.subscribe((req) => {
+    window.localStorage.setItem("SSERequest", JSON.stringify(req))
+  })
+}
+
+function setupMQTTPersistence() {
+  const request = JSON.parse(
+    window.localStorage.getItem("MQTTRequest") || "null"
+  )
+
+  setMQTTRequest(request)
+
+  MQTTRequest$.subscribe((req) => {
+    window.localStorage.setItem("MQTTRequest", JSON.stringify(req))
+  })
+}
+
 function setupGlobalEnvsPersistence() {
   const globals: Environment["variables"] = JSON.parse(
     window.localStorage.getItem("globalEnv") || "[]"
@@ -246,6 +298,10 @@ export function setupLocalPersistence() {
   setupGlobalEnvsPersistence()
   setupEnvironmentsPersistence()
   setupSelectedEnvPersistence()
+  setupWebsocketPersistence()
+  setupSocketIOPersistence()
+  setupSSEPersistence()
+  setupMQTTPersistence()
 }
 
 /**