rxjs support command and runLayersWith

package-lock.json

@@ -14,6 +14,7 @@
         "lodash": "^4.17.21",
         "mustache": "^4.2.0",
         "reflect-metadata": "^0.1.13",
+        "rxjs": "^7.5.5",
         "typed-reflector": "^1.0.10"
       },
       "devDependencies": {
@@ -5843,6 +5844,19 @@
         "queue-microtask": "^1.2.2"
       }
     },
+    "node_modules/rxjs": {
+      "version": "7.5.5",
+      "resolved": "https://registry.npmjs.org/rxjs/-/rxjs-7.5.5.tgz",
+      "integrity": "sha512-sy+H0pQofO95VDmFLzyaw9xNJU4KTRSwQIGM6+iG3SypAtCiLDzpeG8sJrNCWn2Up9km+KhkvTdbkrdy+yzZdw==",
+      "dependencies": {
+        "tslib": "^2.1.0"
+      }
+    },
+    "node_modules/rxjs/node_modules/tslib": {
+      "version": "2.4.0",
+      "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.4.0.tgz",
+      "integrity": "sha512-d6xOpEDfsi2CZVlPQzGeux8XMwLT9hssAsaPYExaQMuYskwb+x1x7J371tWlbBdWHroy99KnVB6qIkUbs5X3UQ=="
+    },
     "node_modules/safe-buffer": {
       "version": "5.1.2",
       "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz",
@@ -11344,6 +11358,21 @@
         "queue-microtask": "^1.2.2"
       }
     },
+    "rxjs": {
+      "version": "7.5.5",
+      "resolved": "https://registry.npmjs.org/rxjs/-/rxjs-7.5.5.tgz",
+      "integrity": "sha512-sy+H0pQofO95VDmFLzyaw9xNJU4KTRSwQIGM6+iG3SypAtCiLDzpeG8sJrNCWn2Up9km+KhkvTdbkrdy+yzZdw==",
+      "requires": {
+        "tslib": "^2.1.0"
+      },
+      "dependencies": {
+        "tslib": {
+          "version": "2.4.0",
+          "resolved": "https://registry.npmjs.org/tslib/-/tslib-2.4.0.tgz",
+          "integrity": "sha512-d6xOpEDfsi2CZVlPQzGeux8XMwLT9hssAsaPYExaQMuYskwb+x1x7J371tWlbBdWHroy99KnVB6qIkUbs5X3UQ=="
+        }
+      }
+    },
     "safe-buffer": {
       "version": "5.1.2",
       "resolved": "https://registry.npmjs.org/safe-buffer/-/safe-buffer-5.1.2.tgz",

package.json

@@ -63,6 +63,7 @@
     "lodash": "^4.17.21",
     "mustache": "^4.2.0",
     "reflect-metadata": "^0.1.13",
+    "rxjs": "^7.5.5",
     "typed-reflector": "^1.0.10"
   },
   "peerDependencies": {

src/decorators/decorators.ts

@@ -4,6 +4,7 @@ import {
   CommandDefinitionFun,
   CommandLocaleDef,
   CommandOptionConfigWithDescription,
+  CommandReturnType,
   ContextCallbackLayer,
   EventName,
   KoishiCommandDefinition,
@@ -17,7 +18,6 @@ import {
 import 'reflect-metadata';
 import {
   Argv,
-  Awaitable,
   BeforeEventMap,
   Command,
   Dict,
@@ -34,6 +34,7 @@ import {
   applyOptionToCommand,
   registerTemplate,
 } from '../utility';
+import { Observable } from 'rxjs';
 
 // Register method
 
@@ -54,22 +55,16 @@ export const UsePlugin = () => DoRegister.plugin();
 export function UseCommand<D extends string>(
   def: D,
   config?: CommandConfigExtended,
-): TypedMethodDecorator<
-  (...args: any[]) => Awaitable<string | void | undefined>
->;
+): TypedMethodDecorator<(...args: any[]) => CommandReturnType>;
 export function UseCommand<D extends string>(
   def: D,
   desc: string,
   config?: CommandConfigExtended,
-): TypedMethodDecorator<
-  (...args: any[]) => Awaitable<string | void | undefined>
->;
+): TypedMethodDecorator<(...args: any[]) => CommandReturnType>;
 export function UseCommand(
   def: string,
   ...args: [CommandConfigExtended?] | [string, CommandConfigExtended?]
-): TypedMethodDecorator<
-  (...args: any[]) => Awaitable<string | void | undefined>
-> {
+): TypedMethodDecorator<(...args: any[]) => CommandReturnType> {
   const desc = typeof args[0] === 'string' ? (args.shift() as string) : '';
   const config = args[0] as CommandConfigExtended;
   return (obj, key: string, des) => {

src/def/interfaces.ts

@@ -12,6 +12,7 @@ import {
 import type { DefaultContext, DefaultState, ParameterizedContext } from 'koa';
 import type { RouterParamContext } from '@koa/router';
 import { CommandPut } from '../registry';
+import { Observable } from 'rxjs';
 
 export interface Type<T = any> extends Function {
   new (...args: any[]): T;
@@ -208,3 +209,7 @@ export type ContextCallbackLayer<T = any> = (
   ctx: Context,
   cb: ContextFunction<void>,
 ) => T;
+
+export type CommandReturnType = Awaitable<
+  string | void | undefined | Observable<string | void | undefined>
+>;

src/register.ts

@@ -14,12 +14,12 @@ import {
 } from './utility';
 import { DoRegister } from './registry';
 import _ from 'lodash';
+import { isObservable, Observable } from 'rxjs';
 
 export interface DoRegisterResult<T> extends DoRegister.Config {
   key: keyof T & string;
   result: any;
 }
-
 export class Registrar<T = any> {
   constructor(
     private obj: T,
@@ -82,20 +82,34 @@ export class Registrar<T = any> {
     ctx: Context,
     cb: ContextFunction<R>,
     layers: ContextCallbackLayer[],
-  ): Promise<Awaited<R>> {
+  ) {
     const rest = [...layers];
     const layer = rest.pop();
-    return new Promise<any>((resolve) => {
-      layer(ctx, async (nextCtx) => {
-        let result: R;
-        if (!rest.length) {
-          result = await cb(nextCtx);
-        } else {
-          result = await this.runLayersWith(nextCtx, cb, rest);
-        }
-        resolve(result);
-      });
-    });
+    return new Observable<R extends Observable<infer U> ? U : Awaited<R>>(
+      (subscriber) => {
+        layer(ctx, async (nextCtx) => {
+          if (!rest.length) {
+            const result = cb(nextCtx);
+            if (result instanceof Promise) {
+              try {
+                subscriber.next(await result);
+              } catch (e) {
+                subscriber.error(e);
+              }
+            } else if (isObservable(result)) {
+              (result as Observable<any>).subscribe({
+                next: (value) => subscriber.next(value),
+                error: (error) => subscriber.error(error),
+              });
+            } else {
+              subscriber.next(result as any);
+            }
+          } else {
+            this.runLayersWith(nextCtx, cb, rest).subscribe(subscriber);
+          }
+        });
+      },
+    );
   }
 
   runLayers<R>(ctx: Context, cb: ContextFunction<R>, key?: keyof T) {

src/registry/registries/do-register.ts

@@ -13,6 +13,7 @@ import {
   BeforeEventName,
   BeforeEventNameAndPrepend,
   CommandRegisterConfig,
+  CommandReturnType,
   EventName,
   EventNameAndPrepend,
   KoaContext,
@@ -31,6 +32,8 @@ import { CommandPut } from './command-put';
 import { applySelector, generateRenderer } from '../../utility';
 import { Next as KoaNext } from 'koa';
 import { IncomingMessage } from 'http';
+import { isObservable, Observable } from 'rxjs';
+import { sessionRxToPromise } from '../../utility/rxjs-session';
 
 // eslint-disable-next-line @typescript-eslint/no-namespace
 export namespace DoRegister {
@@ -88,7 +91,7 @@ export namespace DoRegister {
     onEvent: PickEventFunction<Events, EventName>;
     beforeEvent: PickEventFunction<BeforeEventMap, BeforeEventName>;
     plugin(): Awaitable<PluginDefinition | undefined>;
-    command(...args: any[]): Awaitable<string | void | undefined>;
+    command(...args: any[]): CommandReturnType;
     route(ctx: KoaContext, Next: KoaNext): any;
     ws(socket: WebSocket, request: IncomingMessage): any;
     formatter: I18n.Formatter;
@@ -197,7 +200,7 @@ export namespace DoRegister {
               view,
             );
           }
-          command.action((argv: Argv, ...args: any[]) => {
+          command.action(async (argv: Argv, ...args: any[]) => {
             const params = data.putOptions.map((o, i) =>
               CommandPut.registry.execute(
                 o,
@@ -207,7 +210,15 @@ export namespace DoRegister {
                 view,
               ),
             );
-            return obj[key](...params);
+            const ret = await obj[key](...params);
+            if (isObservable(ret)) {
+              return sessionRxToPromise(
+                argv.session,
+                ret as Observable<string>,
+              );
+            } else {
+              return ret;
+            }
           });
         }
       }

src/utility/rxjs-session.ts

+import { Session } from 'koishi';
+import { Observable } from 'rxjs';
+
+export function sessionRxToPromise(session: Session, obs: Observable<string>) {
+  return new Promise<string>((resolve, reject) => {
+    let lastValue: string;
+    obs.subscribe({
+      next: async (value) => {
+        if (lastValue && session.send) {
+          await session.send(lastValue);
+        }
+        lastValue = value;
+      },
+      error: async (error) => {
+        if (lastValue && session.send) {
+          await session.send(lastValue);
+        }
+        reject(error);
+      },
+      complete: () => {
+        resolve(lastValue);
+      },
+    });
+  });
+}

tests/register.spec.ts

@@ -12,6 +12,7 @@ import {
 import { App, Command, Next, Session } from 'koishi';
 import { Registrar } from '../src/register';
 import { EventNameAndPrepend } from '../src/def';
+import { of } from 'rxjs';
 
 class SkirtArg {
   @PutArg(0)
@@ -39,8 +40,8 @@ class MyClass {
 
   @UseCommand('echo', 'hi')
   @CommandUsage('foo')
-  async onEcho(@PutOption('content', '-c <content>') content: string) {
-    return `bot: ${content}`;
+  onEcho(@PutOption('content', '-c <content>') content: string) {
+    return of(`bot: ${content}`);
   }
 
   @UseCommand('count <count>')