doneit-web/src/app/module/chat/data/async/socket/message-async.service.ts

import { Injectable } from '@angular/core';
import { MessageLiveDataSourceService } from '../../data-source/message/message-live-data-source.service';
import { IncomingMessageSchema, MessageLocalDataSourceService } from '../../data-source/message/message-local-data-source.service';
import { MessageRemoteDataSourceService } from '../../data-source/message/message-remote-data-source.service';
import { SignalRService } from '../../../infra/socket/signal-r.service';
import { filter } from 'rxjs/operators';
import { InstanceId } from '../../repository/message-respository.service';
import { SafeValidateSchema } from 'src/app/services/decorators/validate-schema.decorator';

@Injectable({
  providedIn: 'root'
})
export class MessageAsyncService {

  constructor(
    private messageRemoteDataSourceService: MessageRemoteDataSourceService,
    private messageLiveDataSourceService: MessageLiveDataSourceService,
    private messageLiveSignalRDataSourceService: SignalRService,
    private messageLocalDataSourceService: MessageLocalDataSourceService
  ) {

    this.messageLiveSignalRDataSourceService.getMessage().pipe(
      filter((message: any) => {
        return !message?.requestId?.startsWith(InstanceId) && message?.requestId
      })
    ).subscribe(async (message: any) => {

      console.log('message async ', message)

      const id = message.id + ''
      delete message.id;

      const incomingMessage =  {
        ...message,
        messageId: id,
        sending: false,
        roomId:message.chatRoomId
      }

      this.incomingMessage(incomingMessage)

    })

  }

  @SafeValidateSchema(IncomingMessageSchema, 'socket/incomingMessage')
  async incomingMessage(IncomingMessageSchema: any) {

    const result = await this.messageLocalDataSourceService.sendMessage(IncomingMessageSchema)

    if(result.isOk()) {

    } else {
      console.log(result.error)
    }
  }
}