I have a service that serves a very specific case, gets data from websocket and has to be called recursively to get all results. I’ve managed the service and it looks pretty decent works pretty fast but after trying loads of different ways and even rejigging bits of service, I’ve been unable to write a test for the service.
The service is like so

WebsocketWrapper.ts A websocket wrapper that can be used across system as a shared service and doing it this way means I dont have to import Observables everywhere I need the service and dont have to deal with Observable api, just write a processor method and pass it along as callback

class WebSocketWrapper {
  public subscribeForData({message: ()=>void, error: ()=>void, complete: ()=> void}{
  return socket.subscribe({next: message, error: error, complete: complete};
  }
  public sendMessage(message) {
    this.socket.send(message);
}

Service.ts Needs to query a batch of results and upon receiving and checking response might or might not make request for another set of records

@Injectable()
export class Service implements OnDestroy {
  
  private _allDat = new Subject<MyBigData[]>();
    constructor(
    private _socket: WebSocketWrapper,
  ) {
    // Subscribe to the socket
    this.socketSubscription = this._socket.subscribeForData((resp) => {
      this._funProcess(resp);
    });
  }
  // Socket subscription
  socketSubscription: Subscription;
  // Summary Subscription
  private _allDataSubscription: Subscription | null;
  
  getBatch(conditions): void {
    this._socket.sendMessage({
      msg: 'MyMessageParams'
    });
  }


  private signalEnd = new Subject<void>();
 
  getAllData(recsPerReq: number): Observable<MyBigData[]> {
    const dataSourceObs = new BehaviorSubject<MyBigData[]>([]);

    this._summarySubscription = this._allData.subscribe({
      next: (batch) => {
        dataSourceObs.next(batch);
        const calculatedParams = basedOnWhatWasReturned;
        this.getBatch(
          calculatedParams,
          recsPerReq,
        );
      },
      complete: () => this._logger.info(`completed getting all`),
    });

    this.getBatch(0, recsPerReq);
    
    return dataSourceObs.pipe(
      takeUntil(this.signalEnd),
      reduce((acc, curr) => acc.concat(curr)),
    );
  }
    private _funProcess(respMessage: ResponseType): void {
    // FUn processing
    if (
      conditionsMeat
    ) {
      if (respArray.length == 0) {
        this.signalEnd.next();
      } else {
        this._allData.next(respArray);
      }
    } else if ..
    // Fun process continue
  }
  
  ngOnDestroy(): void {
    // Destroy Subscriptions
  }
}

I’ve changed the test method dramatically quite a few times but nothings worked, stands like this

service.spec.ts Sorry It looks bad now cause I’ve tried quite a few things, will clean up as soon as I can get this to work

beforeEach(() => {
    TestBed.configureTestingModule({
      imports: [LoggerTestingModule],
      providers: [
        Service,
        { provide: WebSocketWrapper, useClass: WebSocketWrapperSpy },
      ],
    });
    jasmine.DEFAULT_TIMEOUT_INTERVAL = 10000;
  });
it('should process a message ', fakeAsync(() => {
    service = TestBed.inject(Service);
    let re: MyBigData[];
    spyOn(service['_allData'], 'next');
    spyOn(service['signalEnd'], 'next');
    spyOn(service, 'getBatch');
    const records = service.getAllData(5);
    tick(501);
    // The following succeeds 
    expect(service['_allData'].next).toHaveBeenCalled();
    tick(5010);
    // This succeeds but fails if i use times(2)
    expect(service.getBatch).toHaveBeenCalled();
    // This fails because the subscirption in the service in all data
    // is never entered and hence, a second call that would've resulted
    // never happens
    expect(service['signalEnd'].next).toHaveBeenCalled();
    // Never gets called
    // I've tried using callback '(done)=> done()' in which case it throws a jasmine
    // default timeout error
    records.subscribe({
      next: (x) => {
        re = x;
        expect(re.length).toEqual(1);
        // done();
      },
      complete: () => console.log(`complete`),
    });
  }));

and finally a websocketSpy
WebSocketSpy.ts A very crude class again, I can do better once it starts to work

class WebSocketServiceSpy {
  private _messageSpy = new Subject<any>();
  private _messageHandler = new Subject<ResponseType>();
  constructor() {
    console.log(`Empty Constructor`);
  }
  messageToSend: ResponseType;
  connect(message: ResponseType): void {}
  public subscribeForBigData(
    process_message: (msg: ResponseType) => void,
    process_error?: (err: any) => void | null,
    process_completion?: () => void | null,
  ): Subscription {
    return this._messageHandler.subscribe({
      next: process_message,
      error: process_error,
      complete: process_completion,
    });
  }
  public sendMessage(msg: SenderTypeMsg): void {

    const resp = {
      msg: 'MyBigData',
      resp: {
        bigData: WebSocketServiceSpy.bigDataResponse.filter(
          (val, idx) =>
            CheckParameterConditions,
        ),
      },
    };
    this._messageHandler.next(resp);
  }
  static bigDataResponse: MyBigData[] = [];
  static bigData: MyBigData = {
    //Big Data Fields
  };
  static {
    WebSocketServiceSpy.bigDataResponse.push(WebSocketServiceSpy.bigData);
  }
}

This service get’s all data, in the caller subscribing to the getAllData() results in all the data.
The test however does not work, if I wanted to test the Response to getAllData().
I suspect, for whatever reason the method executes and finishes and the ‘next’ in subscribe inside getAllData is never reached, at least that’s what I see if I log out to the console,
and I’ve been unable to test the records returned from getAllData().
Any help will be appreciated.
Thanks,