api for events, api test

packages/simplex-chat-client/typescript/tests/client.test.ts

@@ -25,7 +25,7 @@ describe.skip("ChatClient (expects SimpleX Chat server with a user, without cont
     const r2 = await c.msgQ.dequeue()
     assert.strictEqual(r1.type, "contactConnecting")
     assert.strictEqual(r2.type, "contactConnected")
-    const contact1 = (r1 as CEvt.ContactConnected).contact
+    const contact1 = (r1 as CEvt.ContactConnecting).contact
     // const contact2 = (r2 as C.CRContactConnected).contact
     const r3 = await c.apiSendTextMessage(T.ChatType.Direct, contact1.contactId, "hello")
     assert(r3[0].chatItem.content.type === "sndMsgContent" && r3[0].chatItem.content.msgContent.text === "hello")

packages/simplex-chat-nodejs/src/api.ts

@@ -1,4 +1,4 @@
-import {CC, ChatResponse, T} from "@simplex-chat/types"
+import {CC, CEvt, ChatEvent, ChatResponse, T} from "@simplex-chat/types"
 import * as core from "./core"
 
 export class ChatCommandError extends Error {
@@ -12,23 +12,121 @@ export enum ConnReqType {
   Contact = "contact",
 }
 
+export type EventSubscriberFunc<K extends CEvt.Tag> = (event: ChatEvent & {type: K}) => Promise<void>
+
+interface EventSubscriber<K extends CEvt.Tag> {
+  subscriber: EventSubscriberFunc<K>
+  once: boolean
+}
+
 export class ChatApi {
+  private receiveEvents = false
+  private eventsLoop: Promise<void> | undefined = undefined
+  private subscribers: {[K in CEvt.Tag]?: EventSubscriber<K>[]} = {}
+  
   private constructor(protected ctrl_: bigint | undefined) {}
 
-  static async init(dbPath: string, dbKey: string, confirm = core.MigrationConfirmation.YesUp): Promise<ChatApi> {
+  static async init(
+    dbPath: string,
+    dbKey: string = "",
+    confirm = core.MigrationConfirmation.YesUp
+  ): Promise<ChatApi> {
     const ctrl = await core.chatMigrateInit(dbPath, dbKey, confirm)
     return new ChatApi(ctrl)
   }
+    
+  async startChat(): Promise<void> {
+    this.receiveEvents = true
+    this.eventsLoop = this.runEventsLoop()
+    await this.sendChatCmd("/_start")
+    // if (r.type !== "chatStarted") throw new ChatCommandError("error starting chat", r)
+  }
   
+  async stopChat(): Promise<void> {
+    await this.sendChatCmd("/_stop")
+    // if (r.type !== "chatStopped") throw new ChatCommandError("error starting chat", r)
+    this.receiveEvents = false
+    this.eventsLoop = undefined    
+  }
+
   async close(): Promise<void> {
+    this.receiveEvents = false
+    this.eventsLoop = undefined    
     await core.chatCloseStore(this.ctrl)
     this.ctrl_ = undefined
   }
+  
+  private async runEventsLoop(): Promise<void> {
+    while (this.receiveEvents) {
+      try {
+        const event = await this.recvChatEvent()
+        if (!event) continue
+        const subs = this.subscribers[event.type]
+        if (!subs) continue
+        let i = 0;
+        while (i < subs.length) {
+          const {subscriber, once} = subs[i]
+          try {
+            await (subscriber as (event: ChatEvent) => Promise<void>)(event)
+          } catch(e) {
+            console.log(`${event.type} subsriber error`, e)
+          }
+          if (once) subs.splice(i, 1)
+          else i++
+        }
+      } catch(e) {
+        console.log("invalid event", e)
+      }
+    }
+  }
 
+  on<K extends CEvt.Tag>(event: K, subscriber: EventSubscriberFunc<K>, once: boolean = false) {
+    const subs: EventSubscriber<K>[] = this.subscribers[event] || (this.subscribers[event] = [])
+    if (!subs.some(s => s.subscriber === subscriber)) {
+      subs.push({subscriber, once})
+    }
+  }
+  
+  once<K extends CEvt.Tag>(event: K, subscriber: EventSubscriberFunc<K>) {
+    this.on(event, subscriber, true)
+  }
+  
+  wait<K extends CEvt.Tag>(event: K, predicate: ((event: ChatEvent & {type: K}) => boolean) | undefined = undefined): Promise<ChatEvent & {type: K}> {
+    if (predicate) {
+      return new Promise(resolve => {
+        const subscriber: EventSubscriberFunc<K> = async (evt: ChatEvent & {type: K}) => {
+          if (predicate(evt)) {
+            this.off(event, subscriber)
+            resolve(evt)
+          }
+        }
+        this.on(event, subscriber)
+      })
+    } else {
+      return new Promise(resolve => this.once(event, resolve as EventSubscriberFunc<K>))
+    }
+  }
+    
+  off<K extends CEvt.Tag>(event: K, subscriber: EventSubscriberFunc<K> | undefined = undefined) {
+    if (subscriber) {
+      const subs = this.subscribers[event]
+      if (subs) {
+        const i = subs.findIndex(s => s.subscriber === subscriber)
+        if (i !== -1) subs.splice(i, 1)
+      }
+    } else {
+      delete this.subscribers[event]
+    }
+  }
+  
   get initialized(): boolean {
     return typeof this.ctrl_ === "bigint"
-  }  
-  
+  }
+
+  get receiving(): boolean {
+    return this.receiveEvents && this.eventsLoop !== undefined 
+  }
+
   get ctrl(): bigint {
     if (typeof this.ctrl_ === "bigint") return this.ctrl_
     else throw Error("chat api controller not initialized")
@@ -38,6 +136,10 @@ export class ChatApi {
     return await core.chatSendCmd(this.ctrl, cmd)
   }
   
+  async recvChatEvent(wait: number = 15_000_000): Promise<ChatEvent | undefined> {
+    return await core.chatRecvMsgWait(this.ctrl, wait)
+  }
+  
   // Address commands
   // Bots can use these commands to automatically check and create address when initialized
 

packages/simplex-chat-nodejs/src/core.ts

@@ -19,6 +19,7 @@ export async function chatCloseStore(ctrl: bigint): Promise<void> {
 export async function chatSendCmd(ctrl: bigint, cmd: string): Promise<ChatResponse> {
   const res = await simplex.chat_send_cmd(ctrl, cmd)
   const json = JSON.parse(res) as APIResult<ChatResponse>
+  // console.log(cmd, json)
   if (typeof json.result === 'object') return json.result
   if (typeof json.error === 'object') throw new ChatAPIError("Chat command error (see chatError property)", json.error as T.ChatError)
   throw new ChatAPIError("Invalid chat command result")
@@ -29,6 +30,7 @@ export async function chatRecvMsgWait(ctrl: bigint, wait: number): Promise<ChatE
   const res = await simplex.chat_recv_msg_wait(ctrl, wait)
   if (res === "") return undefined
   const json = JSON.parse(res) as APIResult<ChatEvent>
+  // console.log(json)
   if (typeof json.result === 'object') return json.result
   if (typeof json.error === 'object') throw new ChatAPIError("Chat event error (see chatError property)", json.error as T.ChatError)
   throw new ChatAPIError("Invalid chat event")  
@@ -149,7 +151,7 @@ export namespace MigrationError {
 
   export interface MEDowngrade extends Interface {
     type: "downgrade"
-    downMigrations: [string]
+    downMigrations: string[]
   }
 
   export interface MigrationError extends Interface {
@@ -181,6 +183,6 @@ export namespace MTRError {
 
   export interface MTREDifferent extends Interface {
     type: "different"
-    downMigrations: [string]
+    downMigrations: string[]
   }
 }

packages/simplex-chat-nodejs/tests/api.test.ts

@@ -0,0 +1,39 @@
+import * as path from "path";
+import * as fs from "fs";
+import {T} from "@simplex-chat/types"
+import {api} from "../src/index";
+
+describe("Core tests", () => {
+  const tmpDir = "./tests/tmp2";
+  const alicePath = path.join(tmpDir, "alice");
+  const bobPath = path.join(tmpDir, "bob");
+
+  beforeEach(() => fs.mkdirSync(tmpDir, {recursive: true}));
+  afterEach(() => fs.rmSync(tmpDir, {recursive: true, force: true}));
+
+  it("should send/receive message", async () => {
+    const a = await api.ChatApi.init(alicePath)
+    const b = await api.ChatApi.init(bobPath)
+    const aliceUser = await a.apiCreateActiveUser({displayName: "alice", fullName: ""})
+    await b.apiCreateActiveUser({displayName: "bob", fullName: ""})
+    await a.startChat()
+    await b.startChat()
+    const link = await a.apiCreateLink(aliceUser.userId)
+    await b.apiConnectActiveUser(link)
+    const bobContact = (await a.wait("contactConnected")).contact
+    expect(bobContact).toMatchObject({profile: {displayName: "bob"}})
+    const aliceContact = (await b.wait("contactConnected")).contact
+    expect(aliceContact).toMatchObject({profile: {displayName: "alice"}})
+    await a.apiSendTextMessage(T.ChatType.Direct, bobContact.contactId, "hello")
+    await b.wait("newChatItems", ({chatItems}) =>
+      chatItems.some(({chatItem}) => chatItem.meta.itemText === "hello"))
+    await b.apiSendTextMessage(T.ChatType.Direct, bobContact.contactId, "hello too")
+    await a.wait("newChatItems", ({chatItems}) =>
+      chatItems.some(({chatItem}) => chatItem.meta.itemText === "hello too"))
+    // await a.stopChat()
+    // await b.stopChat()
+    await new Promise(resolve => setTimeout(resolve, 2000))
+    await a.close()
+    await b.close()
+  }, 10000)
+})

packages/simplex-chat-nodejs/tests/core.test.ts

@@ -6,9 +6,8 @@ describe("Core tests", () => {
   const tmpDir = "./tests/tmp";
   const dbPath = path.join(tmpDir, "simplex_v1");
 
-  beforeEach(() => fs.mkdirSync(tmpDir, { recursive: true }));
-
-  afterEach(() => fs.rmSync(tmpDir, { recursive: true, force: true }));
+  beforeEach(() => fs.mkdirSync(tmpDir, {recursive: true}));
+  afterEach(() => fs.rmSync(tmpDir, {recursive: true, force: true}));
 
   it("should initialize chat controller", async () => {
     const ctrl = await core.chatMigrateInit(dbPath, "key", core.MigrationConfirmation.YesUp);