feat: implement stop and restart functionality for EventVerse, OrapVerse, and TaskVerse; enhance event handling in app.ts

Author: murongg

packages/orap/examples/declarativeDemo/app.ts

@@ -24,7 +24,7 @@ export function startDemo(options: ListenOptions, storeConfig?: any) {
 
   const toKey: ToKeyFn = (from: string, _to: string, _amount: number) => `${from}_${randomStr(4)}`
 
-  orap.event(eventSignalParam.address, eventSignalParam.abi, eventSignalParam.eventName)
+  const event = orap.event(eventSignalParam.address, eventSignalParam.abi, eventSignalParam.eventName)
     .crosscheck({
       store,
       storeKeyPrefix: 'ora-stack:orap:demo:cc:',
@@ -37,8 +37,8 @@ export function startDemo(options: ListenOptions, storeConfig?: any) {
     // event hook, not necessary
     .handle(newEventSignalHook)
 
-    // add a task
-    .task()
+  // add a task
+  event.task()
     .cache(sm)
     .key(toKey)
     .prefix('ora-stack:orap:demo:TransferTask:', 'ora-stack:orap:demo:Done-TransferTask:')
@@ -53,6 +53,8 @@ export function startDemo(options: ListenOptions, storeConfig?: any) {
     .ttl({ taskTtl: 20000, doneTtl: 20000 })
     .handle(handleTask_2)
 
+  event.stop()
+
   // start signal listener
   orap.listen(
     options,

packages/orap/src/beat/event.ts

@@ -20,4 +20,8 @@ export class EventBeat extends EventSignal {
   drop() {
     this.listen(this.subscribeProvider, this.crosscheckProvider)
   }
+
+  stop() {
+    super.stop()
+  }
 }

packages/orap/src/flow/event.test.ts

@@ -1,5 +1,6 @@
 import { ethers } from 'ethers'
-import { beforeEach, describe, expect, it } from 'vitest'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { EventVerse } from '../verse/event'
 import { OrapFlow } from './orap'
 import { EventFlow } from './event'
 
@@ -48,4 +49,11 @@ describe('EventFlow', () => {
     const parentFlow = eventFlow.another()
     expect(parentFlow).toBe(orapFlow)
   })
+
+  it('should stop the EventVerse', () => {
+    const stopFn = vi.fn()
+    vi.spyOn(EventVerse.prototype, 'stop').mockImplementation(stopFn)
+    eventFlow.stop()
+    expect(stopFn).toHaveBeenCalled()
+  })
 })

packages/orap/src/flow/event.ts

@@ -18,6 +18,12 @@ export class EventFlow implements Flow {
   private _subscribeProvider?: Providers
   private _crosscheckProvider?: Providers
 
+  private _verse: EventVerse = new EventVerse(this)
+
+  get verse() {
+    return this._verse
+  }
+
   constructor(
     private parentFlow?: OrapFlow,
     public params?: EventSignalRegisterParams,
@@ -91,4 +97,12 @@ export class EventFlow implements Flow {
   another(): OrapFlow {
     return this.parentFlow!
   }
+
+  stop() {
+    this._verse.stop()
+  }
+
+  restart() {
+    this._verse.restart()
+  }
 }

packages/orap/src/flow/orap.test.ts

@@ -1,6 +1,7 @@
 import { ethers } from 'ethers'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 import { SEPOLIA_HTTP, SEPOLIA_WSS } from '../../tests/config'
+import { OrapVerse } from '../verse/orap'
 import { OrapFlow } from './orap'
 
 describe('OrapFlow', () => {
@@ -33,4 +34,11 @@ describe('OrapFlow', () => {
     expect(eventFlow).toBeDefined()
     expect(orapFlow.eventFlows).toContain(eventFlow)
   })
+
+  it('should stop the OrapVerse', () => {
+    const stopFn = vi.fn()
+    vi.spyOn(OrapVerse.prototype, 'stop').mockImplementation(stopFn)
+    orapFlow.stop()
+    expect(stopFn).toHaveBeenCalled()
+  })
 })

packages/orap/src/flow/orap.ts

@@ -27,6 +27,12 @@ export class OrapFlow implements Flow {
     return this.subflows.event
   }
 
+  private _verse: OrapVerse = new OrapVerse(this)
+
+  get verse() {
+    return this._verse
+  }
+
   event(params: EventSignalRegisterParams, handler?: HandleFn): EventFlow
   event(address: ContractAddress, abi: Interface | InterfaceAbi | HandleFn, eventName: string, handler?: HandleFn): EventFlow
   event(params: EventSignalRegisterParams | ContractAddress, abi?: Interface | InterfaceAbi | HandleFn, eventName?: string, handler?: HandleFn): EventFlow {
@@ -54,15 +60,23 @@ export class OrapFlow implements Flow {
 
     if (onListenFn)
       this.onListenFn = onListenFn
+    const eventVerses = this.subflows.event.map(flow => flow.verse)
+    this._verse.setEventVerses(eventVerses)
 
-    const orapVerse = this.assemble()
-    orapVerse.play()
+    this._verse.play()
     this.onListenFn()
   }
 
+  stop() {
+    this._verse.stop()
+  }
+
+  restart() {
+    this._verse.restart()
+  }
+
   assemble(): OrapVerse {
     const eventVerses = this.subflows.event.map(flow => flow.assemble())
     return new OrapVerse(this).setEventVerses(eventVerses)
-    // this.routes.event.push(es)
   }
 }

packages/orap/src/flow/task.ts

@@ -16,6 +16,11 @@ const defaultHandleFn: HandleFn = () => {
 }
 
 const defaultToKeyFn: ToKeyFn = _ => randomStr(8, alphabetHex)
+
+const defaultFailFn: HandleResultFn = async (task: TaskRaplized) => {
+  await task.remove()
+}
+
 export interface TaskFlowTTL { taskTtl: Milliseconds; doneTtl: Milliseconds }
 
 export interface TaskFlowParams {
@@ -41,15 +46,17 @@ export class TaskFlow implements Flow {
   toKeyFn: ToKeyFn = defaultToKeyFn
   handleFn: HandleFn = defaultHandleFn
   successFn: HandleResultFn = defaultSuccessFn
+  failFn: HandleResultFn = defaultFailFn
 
   private _middlewares: Array<HandleFn> = []
-
-  failFn: HandleResultFn = async (task: TaskRaplized) => {
-    await task.remove()
-  }
+  private _verse: TaskVerse = new TaskVerse(this)
 
   ctx?: Context
 
+  get verse() {
+    return this._verse
+  }
+
   constructor(
     private parentFlow: EventFlow,
     params?: TaskFlowParams,
@@ -136,4 +143,12 @@ export class TaskFlow implements Flow {
   assemble(): TaskVerse {
     return new TaskVerse(this)
   }
+
+  stop() {
+    this._verse.stop()
+  }
+
+  restart() {
+    this._verse.restart()
+  }
 }

packages/orap/src/signal/event.test.ts

@@ -156,4 +156,14 @@ describe('EventSignal', () => {
       expect(eventSignal.crosschecker?.start).toHaveBeenCalledWith(eventSignal.crosscheckerOptions)
     })
   })
+
+  describe('stop', () => {
+    it('should stop the event listener and the crosschecker', () => {
+      eventSignal.stopEventListener = vi.fn()
+      eventSignal.stopCrossChecker = vi.fn()
+      eventSignal.stop()
+      expect(eventSignal.stopEventListener).toHaveBeenCalled()
+      expect(eventSignal.stopCrossChecker).toHaveBeenCalledWith()
+    })
+  })
 })

packages/orap/src/signal/event.ts

@@ -126,6 +126,23 @@ export class EventSignal implements Signal {
     }
   }
 
+  stop() {
+    this.stopEventListener()
+    this.stopCrossChecker()
+  }
+
+  stopEventListener() {
+    if (this.provider instanceof RekuProviderManager)
+      this.provider.destroy()
+
+    else
+      this.contract.removeListener(this.params.eventName, this.subscribeCallback)
+  }
+
+  stopCrossChecker() {
+    this.crosschecker?.stop()
+  }
+
   async startCrossChecker(provider?: Providers) {
     if (this.crosscheckerParams?.disabled)
       return

packages/orap/src/verse/event.test.ts

@@ -73,4 +73,24 @@ describe('EventVerse', () => {
 
     expect(eventBeatDrop).toHaveBeenCalled()
   })
+
+  it('should stop event verse', () => {
+    const eventBeatStop = vi.fn()
+    vi.spyOn(EventBeat.prototype, 'stop').mockImplementation(eventBeatStop)
+
+    // Mock the _play method to avoid actual EventBeat creation
+    vi.spyOn(eventVerse as any, '_play').mockImplementation(() => {
+      // Create a mock eventBeat
+      (eventVerse as any).eventBeat = {
+        stop: eventBeatStop,
+      }
+    })
+
+    // Call play first to create the eventBeat
+    eventVerse.play()
+    // Then call stop
+    eventVerse.stop()
+
+    expect(eventBeatStop).toHaveBeenCalled()
+  })
 })