RxJ:实现Chatbot延迟消息

Question

我正在开发聊天应用程序，我需要一个接一个地显示机器人消息，在它们之间有一些延迟，这样会给人一种印象，即bot正在键入而不是将所有消息放在一起。我正在用RxJS尝试这种行为，但无法实现所需的输出。

Stackblitz链路

query(): Observable<IChatResponse> {
    const response = {
      messages: [
        {
          text: 'Please give a valid domain name',
        },
        {
          text: 'What domain do you want?',
        },
        {
          text: 'Some other messages...',
        },
      ],
    };

    return of(response).pipe(
      switchMap((response: any) => this.convertToStream(response))
    );
  }

  convertToStream(data: any): Observable<IChatResponse> {
    let count = 0;
    const messageDelayFn = (sme, idx): Observable<any> => {
      const loaderStart$ = of(null).pipe(
        tap((_) => console.log('idx ', idx)),
        delay(500 * idx),
        tap((_) => {
          this.loading$.next(true);
        })
      );
      const loaderStop$ = of(null).pipe(
        delay(1000 * idx),
        tap((_) => {
          this.loading$.next(false);
        })
      );

      const message$ = of(sme);
      return concat(loaderStart$, loaderStop$, message$).pipe(share());
    };

    const transformedObservable = of(data).pipe(
      map((chat) => {
        return {
          ...chat,
          messages: chat.messages.reduce((acc: Observable<any>[], message) => {
            return [...acc, messageDelayFn(message, ++count)];
          }, []),
        };
      })
    );

    return transformedObservable;
  }

我期待的行为是这样的--

• 从500 for的装载机开始
• 装载机停止
• 发出第一条消息
• 加载程序再次启动500 for (在发出第二条消息之前)
• 装载机停止
• 发出第二条讯息
• ...and等

Answer 1

我认为，如果您以不同的方式考虑您的“状态”，可以大大简化您的代码。与单独管理“加载”和“消息”不同，我们可以将它们作为单个接口的一部分来考虑：

export interface ChatBotState {
  messages: string[];
  isTyping: boolean;
}

让我们创建一个可以观察到的状态，以您想要的方式发出这种状态。

我们将首先从查询开始，获取所有消息并一次发出一个消息，因此我们有一个单独的消息流：

  private chatBotMessage$ = this.appService.query().pipe(
    switchMap(response => response.messages)
  );

然后，我们可以使用concatMap / of发出不同的ChatBotStates，并使用所需的时间：

  public state$: Observable<ChatBotState> = this.chatBotMessage$.pipe(
    concatMap(message => concat(
      of({ isTyping: true,  message: undefined    }).pipe(delay(0)),
      of({ isTyping: false, message: message.text }).pipe(delay(500)),
    )),
    // ... more to come

在这里，我们使用isTyping = true立即发出一个状态，然后在500ms之后发出一个带有消息文本和isTyping = false的状态。

我们可以使用scan发出包含所有消息的结果状态对象：

const INITIAL_STATE: ChatBotState = {
  messages: [],
  isTyping: false,
};

  public state$: Observable<ChatBotState> = this.chatBotMessage$.pipe(
    concatMap(message => concat(
      of({ isTyping: true,  message: undefined    }).pipe(delay(0)),
      of({ isTyping: false, message: message.text }).pipe(delay(1500)),
    )),
    scan((previous, current) => ({
      isTyping: current.isTyping,
      messages: current.message 
                  ? previous.messages.concat(current.message) 
                  : previous.messages
    }), INITIAL_STATE)
  );

下面是一个工作的StackBlitz演示。

由于state$可观测到的数据正是视图所需的，所以我们的组件变得非常简单！不需要管理订阅，所以不需要ngOnInit或ngOnDestroy。您可以使用模板中的单个async管道来打开视图模型，仅此而已！

export class AppComponent {

  vm$ = this.chatBotService.state$;

  constructor(private chatBotService: ChatBotService) { }

}

<div *ngIf="vm$ | async as vm">
  
  <p *ngFor="let message of vm.messages">
    {{ message }}
  </p>

  <div *ngIf="vm.isTyping">
    Loading...
  </div>

</div>

Answer 2

这是一个符合您的规范的函数。您可以自己尝试运行它，然后希望能够根据需要将它调整到您的用例中。

function query(): Observable<string> {
  const response = {
    messages: [
      {
        text: 'Please give a valid domain name',
      },
      {
        text: 'What domain do you want?',
      },
      {
        text: 'Some other messages...',
      },
    ],
  };

  return concat(...response.messages.map(({text}) => 
    timer(500).pipe(map(_ => text))
  ));
}

query().subscribe(console.log);

Answer 3

服务

export default class AppService {
  query(): Observable<IMessage> {
    const response = {
      messages: [
        {
          text: 'Please give a valid domain name',
        },
        {
          text: 'What domain do you want?',
        },
        {
          text: 'Some other messages...',
        },
      ],
    };
    return from(response.messages);
  }
}

组件

export class AppComponent {
  text$ = this.appService.query().pipe(
    concatMap(({ text }) => {
      const share$ = of(text).pipe(delay(1000), share());
      return share$.pipe(delay(1000), startWith(share$), skipLast(1));
    }),
    scan((arr, v) => [...arr, v], [])
  );
  constructor(private appService: AppService) {}
}

html

<div *ngFor="let item of text$ | async">
  <div class="text">
    <span *ngIf="item | async as item; else loading">
      {{ item }}
    </span>
    <ng-template #loading>
      <img src="https://c.tenor.com/VS20soWAM9AAAAAi/loading.gif" width="30"/>
    </ng-template>
  </div>
</div>

https://stackblitz.com/edit/angular-ivy-ot6dcr