仅在成功连接后才调用SignalR

Question

我在做一个角质水疗。总的来说，我对棱角，打字和网页开发都很陌生，所以请原谅我。

该网页包含一个“职务”列表。用户可以打开作业的详细信息，调整参数，并告诉服务器执行此作业。当用户打开作业时，让我们称其为作业详细信息组件，SignalR用于发送有关此作业细节的更新。这通过作业细节组件中的ngOnInit中的invoke命令实现。

app-component:

ngOnInit() {
  this.SignalRService.startConnection();
}

job-detail-component:

job: IJob;

ngOnInit() {    
    var jobId = this.route.snapshot.paramMap.get('jobId');
    this.SignalRService.hubConnection.on('jobstate', (data: any) => {
      console.log("received job state");
      this.job = data;
    });
    this.SignalRService.subscribeJob(jobId);

SignalR-service

public async startConnection() {
  this.hubConnection.start()
    .then(() => {
       console.log("Connected to server");
      })
    .catch(err => {
       console.log('Error while starting connection: ');
       console.log(err);
    }
  )
}

public subscribeJob(jobId: string) {
  console.log("subscribing to job");
  this.hubConnection.invoke("SubJobState", jobId);
}

http://localhost:5000/job-detail/1说，现在上面的部分很好用，直到有人直接通过与工作的直接链接进入工作。现在，它们输入了作业编号1的详细信息页，hubconnection.start()由app.component.ts执行，hubconnection.invoke()从作业详细组件执行，并引发以下错误：

Error: Uncaught (in promise): Error: Cannot send data if the 
connection is not in the 'Connected' State.

这必须发生，因为SignalRService.startConnection()目前正在连接到集线器，并且在创建连接之前调用调用。

我解决了这个问题，我的意思可能是一个非理想的解决方案：

public subscribeJob(jobId: string) {
  if (this.hubConnection.state === 1) {
    console.log("subscribing to job");
    this.hubConnection.invoke("SubJobState", jobId);
  }
  else {
    setTimeout(() => {
      this.subscribeJob(jobId);
    }, 500);
  }
}

有比这种方法更好的解决方法吗？

Answer 1

下面是一个允许您随时调用invoke的示例代码。

import {Injectable} from "@angular/core";
import {HubConnection, HubConnectionBuilder} from '@microsoft/signalr';
import {from, Observable, Subject} from "rxjs";
import {mergeMap} from "rxjs/operators";

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

  private _hubConnection: HubConnection;
  private readonly startedConnection:Promise<any>;

  constructor() {
    this._hubConnection = new HubConnectionBuilder()
      .withUrl(`${window.location.href}chatHub`)
      .build();

    this.startedConnection = this.startConnection();
  }

  on(method: string): Observable<any> {

    const subject = new Subject<any>();

    this._hubConnection.on(method, data => {
      return subject.next(data);
    });

    return subject.asObservable();
  }


   invoke(method: string, ...data: any): Observable<any> {

     return from(this.startedConnection)
       .pipe(
         mergeMap(_ => {
           return this._hubConnection.invoke(method, ...data)
         })
       )

  }

  private startConnection() {
    return this._hubConnection
      .start()
      .then(() => {
        console.log('Hub connection started');
        return 'connected'
      })
      .catch(err => {
        console.log('Error while establishing connection, retrying...');
        setTimeout(this.startConnection, 5000)
      });
  }

}

将其插入组件并按以下方式使用：

export class HomeComponent implements  OnDestroy{

  private  subscriptions = new Subscription();

  constructor(private signalr: SignalrService) {}

  users$ = this.signalr.on("ReceiveUsers");
  messages$ = this.signalr.on("ReceiveMessages");


  ngOnInit(): void {
    const sendUsersSub = this.signalr.invoke("SendUsers").subscribe();
    this.subscriptions.add(sendUsersSub);

  }

  ngOnDestroy(): void {
    this.subscriptions.unsubscribe();
  }

}

Answer 2

我曾经遇到过这个问题，我就这样解决了它。

想象职能A是一种公共职能，它：

1-可以随时随地调用。(在任何组件中)

2-需要创建SingnalR连接。

    public A() {
        //some code that needs SignalR connection to be created
    }

通过开发项目，您将添加一些类似于函数A的其他函数。

     public B() {
         // like function A , it needs SignalR connection to be created
      }

与其随时随地调用这两个函数，不如调用一个公共的manager函数，并接收一个值(可以是enum)，该值告诉我们应该调用哪个先前的函数。另外，让A和B私人化。

    private A() {
        //some code that needs SignalR connection to be created
    }

    private B() {
         // like function A , it needs SignalR connection to be created
    }

    public manager(type:string) {
        if (type === 'A') this.A();
        else if (type === 'B') this.B();
    }

那么为什么要添加管理器函数呢？因为您可以将其连接到控制连接状态的布尔变量(让我们称之为已连接)。如果连接是真的，那么运行它。如果连接不是真，则可以将其添加到队列中。为了单一的责任，我只检查经理函数中的连接状态和其他职责移交给其他两个功能。

    public manager(type:string) {
        if (this.connected === true) this.callProperFunction(type)
        else this.addToQueue(type)
    }
    private callProperFunction(type:string) {
        if (type === 'A') this.A();
        else if (type === 'B') this.B();
    }
    private addToQueue(type:string) {
        this.queue.push(type);
        // you can implement a queue with simple array.
        // just add to end (push) and read from front (shift)
    }

最后一部分是关于变量(连接:布尔)，它必须侦听signalR的状态。您可以很容易地实现它的角度可观察。

    this.connection.subscribe((state:boolean) => {
          //first change the state for future calling
        this.connected = state; 
          //run all the functions which was store in queue.
        for (let i=0; i<this.queue.length; i++) {
          //call from the front
          const type = this.queue.shift();
          //now call all the methods  which was blocked when the state                          
          // was false and not connected 
          this.callProperFunction(type);
        }
    }

最后注意:无论何时创建signalR连接，您都必须告诉所有订阅者！

private startConnection() {
    return this._hubConnection
      .start()
      .then(() => {
        console.log('Hub connection started');
        **this.someSharedService.changeConnectionState(true);**
        return 'connected'
      })
      .catch(err => {
        console.log('Error while establishing connection, retrying...');
        setTimeout(this.startConnection, 5000)
      });
  }