Minimal Angular WAMP Service

wamp.service.ts

import { Injectable } from '@angular/core';
import { Connection, Session } from 'autobahn-browser';
import { Observable, Observer, ReplaySubject } from 'rxjs';
import { switchMap } from 'rxjs/operators';

const wampUrl = '<YOUR_WAMP_SERVER_URL>';
const wampRealm = '<YOUR_REALM_NAME>';

@Injectable({ providedIn: 'root' })
export class WampService {
  private session$ = new ReplaySubject<Session>(1);

  constructor() {
    const connection = new Connection({
      url: wampUrl,
      realm: wampRealm
      // Implement authentication if needed
    });
    connection.onopen = session => {
      console.log('Successfully connected', wampUrl);
      this.session$.next(session);
    };
    connection.open();
  }

  public listen(topicUri: string): Observable<any> {
    return this.session$.pipe(
      switchMap(session => {
        return new Observable((subscriber: Observer<any>) => {
          session.subscribe(topicUri, event => subscriber.next(event));
        });
      })
    );
  }

  public send(topic: string, arr: any[], obj?: object): void {
    this.session$.subscribe(session => {
      session.publish(topic, arr, obj);
    });
  }

  public call(procUri: string, arr?: any[], obj?: object, options?: object): Observable<any> {
    return this.session$.pipe(
      switchMap(session => {
        return new Observable((subscriber: Observer<any>) => {
          session
            .call(procUri, arr, obj, options)
            .then(subscriber.next.bind(subscriber))
            .catch(subscriber.error.bind(subscriber))
            .finally(subscriber.complete.bind(subscriber));
        });
      })
    );
  }

  public register(
    procName: string,
    proc: (arr: any[], obj?: object, options?: object) => any
  ): Observable<any> {
    return this.session$.pipe(
      switchMap(session => {
        return new Observable((subscriber: Observer<any>) => {
          session
            .register(procName, proc)
            .then(subscriber.next.bind(subscriber))
            .catch(subscriber.error.bind(subscriber))
            .finally(subscriber.complete.bind(subscriber));
        });
      })
    );
  }
}