receive user typing

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

@@ -5,7 +5,6 @@ import { MessageRemoteDataSourceService } from '../../data-source/message/messag
 import { SignalRService } from '../../../infra/socket/signal-r.service';
 import { filter } from 'rxjs/operators';
 import { InstanceId } from '../../repository/message-respository.service';
-import { SocketStreamReturn } from 'src/app/services/decorators/socket-validate-schema.decorator';
 import { SafeValidateSchema } from 'src/app/services/decorators/validate-schema.decorator';
 
 @Injectable({

src/app/module/chat/data/async/socket/room-async.service.ts

@@ -0,0 +1,36 @@
+import { Injectable } from '@angular/core';
+import { MessageLiveDataSourceService } from '../../data-source/message/message-live-data-source.service';
+import { RoomLiveDataSourceService } from '../../data-source/room/room-live-data-source.service';
+import { RoomRemoteDataSourceService } from '../../data-source/room/room-remote-data-source.service';
+import { roomDataSource, RoomLocalDataSourceService } from '../../data-source/room/rooom-local-data-source.service';
+
+@Injectable({
+  providedIn: 'root'
+})
+export class RoomAsyncService {
+
+  constructor(
+    private roomRemoteDataSourceService: RoomRemoteDataSourceService,
+    // private roomMemoryDataSourceService: Store<RoomRemoteDataSourceState>,
+    private roomLocalDataSourceService: RoomLocalDataSourceService,
+    private roomLiveDataSourceService: RoomLiveDataSourceService,
+    private messageLiveDataSourceService: MessageLiveDataSourceService,
+  ) {
+
+
+    roomDataSource.typing.hook('creating', (primKey, obj, trans) => {
+      setTimeout(() => {
+
+      }, 1000);
+    })
+  }
+
+
+  incomingTyping() {
+
+  }
+
+  async removeUserTyping() {
+    const result = await this.roomLocalDataSourceService.removeUserTyping()
+  }
+}

src/app/module/chat/data/async/socket/user-typing-async.service.ts

@@ -0,0 +1,53 @@
+import { Injectable } from '@angular/core';
+import { UserTypingLiveDataSourceService } from '../../data-source/userTyping/user-typing-live-data-source.service';
+import { UserTypingLocalDataSourceService } from '../../data-source/userTyping/user-typing-local-data-source.service';
+import { SignalRService } from '../../../infra/socket/signal-r.service';
+import { interval, Subject, timer } from 'rxjs';
+import { switchMap } from 'rxjs/operators';
+import { addUserTyping, removeUserTyping, TypingState } from '../../data-source/userTyping/user-typing-memory-data-source.service';
+import { Store } from '@ngrx/store';
+
+@Injectable({
+  providedIn: 'root'
+})
+export class UserTypingAsyncService {
+
+  typingCallback: {[key: string]: Subject<any> } = {}
+
+  constructor(
+    private localDataSource: UserTypingLocalDataSourceService,
+    private liveDataSource: UserTypingLiveDataSourceService,
+    private memoryDataSource: Store<TypingState>,
+    private signalR: SignalRService,
+  ) {
+
+    this.signalR.getTyping().subscribe(async (e:any) => {
+      if(e?.chatRoomId) {
+
+        console.log('e', e)
+
+        this.memoryDataSource.dispatch(removeUserTyping({data: {...e} as any}))
+        this.memoryDataSource.dispatch(addUserTyping({data: {...e} as any}))
+        //
+        const value = await this.localDataSource.addUserTyping(e);
+
+        const id =  e.chatRoomId + '@' + e.userName
+        if(!this.typingCallback[id]) {
+          this.typingCallback[id] = new Subject()
+          this.typingCallback[id].pipe(
+            switchMap(() => timer(2000)),
+          ).subscribe(() => {
+            console.log('111111==============')
+            this.memoryDataSource.dispatch(removeUserTyping({data: {...e} as any}))
+            this.localDataSource.removeUserTyping(e)
+          })
+        } else {
+          this.typingCallback[id].next()
+        }
+
+      } else {
+        console.log('e--', e)
+      }
+    })
+  }
+}