UniRX/Reactive Extensions: Signal into Queue-Signal/Overlapping Buffer

Hey, I thought this Question might be too specific for Unity Answers so I'm asking here:

With UniRX, I would like to convert a signal emitting values into a signal emitting lists of these values, but not as a buffer but rather a kind of Queue: From an Observable emitting numbers (let's say floats), I want to generate an Observable always emitting a List of the three most recently emitted values. So as a RX-Marbles-like graph, I want to convert it like this:

Code (CSharp):

1.  
2. ---1--------1---------2-----------3-----------4-----------1-----------|
3.    |        |         |           |           |           |           |
4. ----------------------List(1,1,2)-List(1,2,3)-List(2,3,4)-List(3,4,1)-|
5.  

It would also be fine if the resulting Signal already started emitting before 3 Values are available, only emitting those (so in this case starting with List(1) and then List(1,1)).

Does any UniRX-User know an operator for that / is there in general an operator for something like that in Reactive Extensions?

If it doesn't extist, an idea I have is taking the original signals, creating two other signals from them and give them an "item-delay" of 1 and 2 (i.e. for example when the 7th item is emitted, one emits the 6th and the other one emits the 5th) and then zipping them all together. But is there an operator for this "item-delay"?

Thank you in advance for any suggestions/hints!

 

Sorry mate, I've never used UniRX before, but considering this is such a niche question it may well go unanswered so I'll stick my two cents in.

Is there any reason why you couldn't keep an array of the three values and add to it like this?:

Code (CSharp):

1.  
2.     float[] values = new float[3];
3.     int currentValue = 0;
4.     void AddValue(float newValue)
5.     {
6.         currentValue++;
7.         if(currentValue >= 3)
8.         {
9.             currentValue = 0;
10.         }
11.  
12.         values[currentValue] = newValue;
13.     }
14.  

I understand it has something to do with asynchronous events, so this is in all likelihood going right over my head.

 

I think you should like at Observable.Buffer maybe? or maybe Observable.Take(3)?
By the way if you are active can you please contact me? I have an issue with UniRx that I do not understand. I would highly appreciate your input

 

Stumbled across this old thread while searching for something unrelated. Figured I'd answer for anyone else who is interested. RX can be tough to wrap your head around, and lots of examples can help.

This is a good use for the Zip operator: http://reactivex.io/documentation/operators/zip.html
Instead of zipping unrelated observables together, we zip the same observable but shift its output.

Code (csharp):

1.  
2. private void Start()
3. {
4.     // create a mock signal
5.     var signal = new Subject<int>();
6.  
7.     var observer = Observable.Zip(
8.         signal,
9.         signal.Skip(1),
10.         signal.Skip(2));
11.    
12.     using (observer.Subscribe(ints => Debug.Log($"{ints[0]}, {ints[1]}, {ints[2]}")))
13.     {
14.         // fire off some signals
15.         signal.OnNext(1);
16.         signal.OnNext(1);
17.         signal.OnNext(2);
18.         signal.OnNext(3);
19.         signal.OnNext(4);
20.         signal.OnNext(1);
21.     }
22. }

Result:
1, 1, 2
1, 2, 3
2, 3, 4
3, 4, 1