Блог

Я использую SockJS и StompJS, и когда я открываю свое приложение в браузере, иногда оно пытается подписаться на некоторые темы, прежде чем оно даже подключится к websocket. Я бы хотел, чтобы подписка на тему подождала, пока приложение не будет подключено к websocket.

 export class SocksService {
...

public subscribe<T>(destination: string, callback?: (body: T) => void, headers?: object): Observable<Subscription> {
const subscription = new Subject<Subscription>();
headers = headers || {};

this.status.pipe(
  first(status => status === ConnectionStatus.CONNECTED)
).subscribe((status: ConnectionStatus) => {
  subscription.next(this.stompClient.subscribe(destination, (message: Message) => {
    if (callback) {
      callback(JSON.parse(message.body) as T);
    }
  }, headers));
});

return subscription.asObservable();
  }

...
}
 

Вот почему я реализовал этот код и называю его как:

 this.socksService.subscribe<User>('/topic/user', (user: User) => {
  console.log('user received', user);
}).subscribe(subscription => this.userSubscription = subscription);
 

Поэтому я подписываюсь на тему только тогда, когда статус соединения есть connected , и он будет вызываться только при первом успешном подключении клиента.

Я хотел бы позже отписаться от темы, поэтому мне нужен Subscription объект, возвращаемый внутренней подпиской, и мне также нужно сообщение от внутренней подписки.

То, что я реализовал, работает хорошо, но я думаю, что должен быть лучший способ сделать это.

(Я пробовал rx-stomp, но в нем много ошибок.)

Вы сказали, что пробовали rx-stomp, но пробовали ли вы Angular2 версию rx-stomp, называемую ng2-stompjs? Он предоставляет ключевые классы как сервисы, которые можно вводить с помощью Angular.

См. Упоминание об этом в разделе rx-stomp, здесь: https://github.com/stomp-js/ng2-stompjs#documentation

… и это реализация (Angular 7) с помощью приятного пошагового руководства, здесь: https://stomp-js.github.io/guide/ng2-stompjs/ng2-stomp-with-angular7.html#receiving-messages

например:

 this.rxStompService.watch('/topic/demo').subscribe((message: Message) => {
      this.receivedMessages.push(message.body);
    });