public get(
   query: Ha4usObjectQuery,
   aTimeout: number = 500
 ): Promise<IStatefulObject[]> {
   return this.$objects
     .observe(query)
     .pipe(
       mergeMap((aObject: Ha4usObject) => {
         return this.$states.observe(aObject.topic).pipe(
           timeoutWith(100, of({})),
           take(1),
           map((state: Ha4usMessage) => {
             delete state.topic
             delete state.match
             return {
               object: aObject,
               state: state,
             }
           })
         )
       }),
       bufferTime(aTimeout),
       timeout(aTimeout * 2)
     )
     .toPromise()
 }

  it('should emit buffers with timeSpan 100 and creationInterval 70', () => {
    const e1 = hot('--1--^2--3---4---5--6--7---8----9------------|');
                    // -------*------*------*------*------*----- creation interval
                    // ----------|                               timespans
                    //        ----------|
                    //               ----------|
                    //                      ----------|
                    //                             ----------|
                    //                                    ----------|
    const e1subs =      '^                                       !';
    const t = time(     '----------|');
    const interval = time(        '-------|');
    const expected =    '----------a------b------c------d------e-(f|)';
    const values = {
      a: ['2', '3', '4'],
      b: ['4', '5', '6'],
      c: ['6', '7', '8'],
      d: ['8', '9'],
      e: [] as string[],
      f: [] as string[]
    };

    const result = e1.pipe(bufferTime(t, interval, Number.POSITIVE_INFINITY, rxTestScheduler));

    expectObservable(result).toBe(expected, values);
    expectSubscriptions(e1.subscriptions).toBe(e1subs);
  });

 bufferTime1() {
   // Create an observable that emits a value every 500ms
   const source = interval(500);
   // After 2 seconds have passed, emit buffered values as an array
   const example = source.pipe(bufferTime(2000));
   // Print values to console
   // ex. output [0,1,2]...[3,4,5,6]
   const subscribe = example.subscribe(val =>
     console.log('Buffered with Time:', val)
   );
 }

 bufferTime2() {
   // Create an observable that emits a value every 500ms
   const source = interval(500);
   /*
     bufferTime also takes second argument, when to start the next buffer (time in ms)
     for instance, if we have a bufferTime of 2 seconds but second argument (bufferCreationInterval) of 1 second:
     ex. output: [0,1,2]...[1,2,3,4,5]...[3,4,5,6,7]
   */
   const example = source.pipe(bufferTime(2000, 1000));
   // Print values to console
   const subscribe = example.subscribe(val =>
     console.log('Start Buffer Every 1s:', val)
   );
 }

 return (actor$: Observable<Actor>) => {
   return actor$.pipe(
     rxFilter((actor) => RequestActor.isPreRequestActor(actor)),
     rxObserveOn(asyncScheduler),
     rxBufferTime(100),
     rxFilter((actors) => actors.length > 0),
     rxMergeMap((actors) => {
       const nextActors = {} as Dictionary<Actor>;
       actors.forEach((actor) => {
         nextActors[actor.type + JSON.stringify(actor.arg)] = actor;
       });
       return values(nextActors);
     }),
   );
 };