add support for contentTopic change

src/app/components/Waku.tsx

@@ -1,13 +1,27 @@
 import React from "react";
-import { Block, BlockTypes } from "@/components/Block";
+import { Block } from "@/components/Block";
 import { Subtitle } from "@/components/Subtitle";
 import { Button } from "@/components/Button";
 import { MessageContent, useWaku } from "@/hooks";
-import { CONTENT_TOPIC } from "@/constants";
 
 export const Waku: React.FunctionComponent<{}> = () => {
-  const { onSend, messages } = useWaku();
+  const {
-  const { nick, text, onNickChange, onMessageChange, resetText } = useMessage();
+    onSend,
+    messages,
+    contentTopic: activeContentTopic,
+    onContentTopicChange: onActiveContentTopicChange
+  } = useWaku();
+  const {
+    nick,
+    text,
+    onNickChange,
+    onMessageChange,
+    resetText,
+  } = useMessage();
+  const {
+    contentTopic,
+    onContentTopicChange,
+  } = useContentTopic(activeContentTopic);
 
   const onSendClick = async () => {
     await onSend(nick, text);
@@ -26,7 +40,20 @@ export const Waku: React.FunctionComponent<{}> = () => {
           <Subtitle>
             Waku
           </Subtitle>
-          <p className="text-sm">Content topic: {CONTENT_TOPIC}</p>
+          <label
+            htmlFor="contentTopic-input"
+            className="block mb-2 mt-2 text-sm font-medium text-gray-900 dark:text-white"
+          >
+            Content topic
+          </label>
+          <input
+            type="text"
+            id="contentTopic-input"
+            value={contentTopic}
+            onChange={onContentTopicChange}
+            className="w-96 mr-2 bg-gray-50 border border-gray-300 text-gray-900 text-sm rounded-lg focus:ring-blue-500 focus:border-blue-500 block p-2.5 dark:bg-gray-700 dark:border-gray-600 dark:placeholder-gray-400 dark:text-white dark:focus:ring-blue-500 dark:focus:border-blue-500"
+          />
+          <Button className="mt-1" onClick={() => { onActiveContentTopicChange(contentTopic); }}>Change</Button>
         </Block>
 
         <Block className="mt-4 mr-10 min-w-fit">
@@ -76,6 +103,23 @@ export const Waku: React.FunctionComponent<{}> = () => {
   );
 };
 
+function useContentTopic(globalContentTopic: string) {
+  const [contentTopic, setContentTopic] = React.useState<string>(globalContentTopic);
+
+  React.useEffect(() => {
+    setContentTopic(globalContentTopic);
+  }, [globalContentTopic]);
+
+  const onContentTopicChange = (e: React.SyntheticEvent<HTMLInputElement>) => {
+    setContentTopic(e.currentTarget.value || "");
+  };
+
+  return {
+    contentTopic,
+    onContentTopicChange,
+  };
+}
+
 function useMessage() {
   const [nick, setNick] = React.useState<string>("");
   const [text, setText] = React.useState<string>("");

src/hooks/useWaku.ts

@@ -5,10 +5,11 @@ import { Message, waku } from "@/services/waku";
 export type MessageContent = {
   nick: string;
   text: string;
-  time: string;
+  timestamp: number;
 };
 
 export const useWaku = () => {
+  const [contentTopic, setContentTopic] = React.useState<string>(CONTENT_TOPIC);
   const [messages, setMessages] = React.useState<Map<string, MessageContent>>(new Map());
 
   React.useEffect(() => {
@@ -19,7 +20,7 @@ export const useWaku = () => {
       newMessages.forEach((m) => {
         const payload = JSON.parse(atob(m.payload));
 
-        const message = {
+        const message: MessageContent = {
           nick: payload?.nick || "unknown",
           text: payload?.text || "empty",
           timestamp: m.timestamp || Date.now(),
@@ -30,12 +31,12 @@ export const useWaku = () => {
       setMessages(nextMessages);
     };
 
-    waku.relay.addEventListener(CONTENT_TOPIC, messageListener);
+    waku.relay.addEventListener(contentTopic, messageListener);
 
     return () => {
-      waku.relay.removeEventListener(CONTENT_TOPIC, messageListener);
+      waku.relay.removeEventListener(contentTopic, messageListener);
     };
-  }, [messages, setMessages]);
+  }, [messages, setMessages, contentTopic]);
 
   const onSend = React.useCallback(
     async (nick: string, text: string) => {
@@ -62,5 +63,18 @@ export const useWaku = () => {
     [setMessages]
   );
 
-  return { onSend, messages: Array.from(messages.values()) };
+  const onContentTopicChange = async (nextContentTopic: string) => {
+    if (nextContentTopic === contentTopic) {
+      return;
+    }
+
+    setContentTopic(nextContentTopic);
+  };
+
+  return {
+    onSend,
+    contentTopic,
+    onContentTopicChange,
+    messages: Array.from(messages.values())
+  };
 };

src/services/waku.ts

@@ -18,16 +18,13 @@ const buildURL = (endpoint: string) => `${LOCAL_NODE}${endpoint}`;
 
 class Relay {
   private readonly subscriptionsEmitter = new EventTarget();
-
-  private contentTopicListeners: Map<string, number> = new Map();
-
   // only one content topic subscriptions is possible now
   private subscriptionRoutine: undefined | number;
 
   constructor() {}
 
   public addEventListener(contentTopic: string, fn: EventListener) {
-    this.handleSubscribed(contentTopic);
+    this.subscribe(contentTopic);
     return this.subscriptionsEmitter.addEventListener(contentTopic, fn as any);
   }
 
@@ -38,12 +35,8 @@ class Relay {
     );
   }
 
-  private async handleSubscribed(contentTopic: string) {
+  private async subscribe(contentTopic: string) {
-    const numberOfListeners = this.contentTopicListeners.get(contentTopic);
+    if (this.subscriptionRoutine) {
-
-    // if nwaku node already subscribed to this content topic
-    if (numberOfListeners) {
-      this.contentTopicListeners.set(contentTopic, numberOfListeners + 1);
       return;
     }
 
@@ -53,22 +46,13 @@ class Relay {
       this.subscriptionRoutine = window.setInterval(async () => {
         await this.fetchMessages();
       },  5 * SECOND);
-
-      this.contentTopicListeners.set(contentTopic, 1);
     } catch (error) {
       console.error(`Failed to subscribe node ${contentTopic}:`, error);
     }
   }
 
-  private async handleUnsubscribed(contentTopic: string) {
+  public async unsubscribe(contentTopic: string) {
-    const numberOfListeners = this.contentTopicListeners.get(contentTopic);
+    if (!this.subscriptionRoutine) {
-
-    if (!numberOfListeners) {
-      return;
-    }
-
-    if (numberOfListeners - 1 > 0) {
-      this.contentTopicListeners.set(contentTopic, numberOfListeners - 1);
       return;
     }
 
@@ -79,16 +63,9 @@ class Relay {
     }
 
     clearInterval(this.subscriptionRoutine);
-    this.contentTopicListeners.delete(contentTopic);
   }
 
   private async fetchMessages(): Promise<void> {
-    const contentTopic = Array.from(this.contentTopicListeners.keys())[0];
-
-    if (!contentTopic) {
-      return;
-    }
-
     const response = await http.get(
       buildURL(`${RELAY}/messages/${encodeURIComponent(PUBSUB_TOPIC)}`)
     );
@@ -98,9 +75,27 @@ class Relay {
       return;
     }
 
+    const messagesPerContentTopic = new Map<string, Message[]>();
+    body.forEach((m) => {
+      const contentTopic = m.contentTopic;
+      if (!contentTopic) {
+        return;
+      }
+
+      let messages = messagesPerContentTopic.get(contentTopic);
+      if (!messages) {
+        messages = [];
+      }
+
+      messages.push(m);
+      messagesPerContentTopic.set(contentTopic, messages);
+    });
+
+    Array.from(messagesPerContentTopic.entries()).forEach(([contentTopic, messages]) => {
       this.subscriptionsEmitter.dispatchEvent(
-      new CustomEvent(contentTopic, { detail: body })
+        new CustomEvent(contentTopic, { detail: messages })
       );
+    });
   }
 
   public async send(message: Message): Promise<void> {