zip 3 or more streams

[xiaojian-hong]

I thought it would be straightforward, but I'm facing some difficulties...

I have a data source app that sends 5 packets encoded with 5 different keys:

func generateAndSendData(stream quic.Stream) {
    keys := []byte{0x10, 0x11, 0x13, 0x7E, 0x7F}

    for i, key := range keys {
        time.Sleep(100 * time.Millisecond)
        
        codec := y3.NewCodec(key)

        sendingBuf, _ := codec.Marshal(int64(i))

        _, err := stream.Write(sendingBuf)
        if err != nil {
            log.Printf("Couldn't send buffer with i=%v", i)
        } else {
            fmt.Print(".")
        }
    }
}

And the flow app that has 5 streams that are subscribed to each individual key and then zipped together:

var zipper = func(_ context.Context, a interface{}, b interface{}) (interface{}, error) {
    accumulator, ok := a.([]interface{})
    if !ok {
        fmt.Printf("No accumulator: %v + %v\n", a, b)
        return []interface{}{a, b}, nil
    }

    fmt.Printf("With accumulator: %v + %v\n", accumulator, b)
    accumulator = append(accumulator, b)
    return accumulator, nil
}

var convert = func(v []byte) (interface{}, error) {
    fmt.Printf("Got: %v\n", v)
	return y3.ToInt64(v)
}

// Handler will handle data in Rx way
func Handler(rxstream rx.RxStream) rx.RxStream {
	streamA:= rxstream.Subscribe(0x10).OnObserve(convert)
	streamB:= rxstream.Subscribe(0x11).OnObserve(convert)
	streamC:= rxstream.Subscribe(0x13).OnObserve(convert)
	streamD:= rxstream.Subscribe(0x7E).OnObserve(convert)
	streamE:= rxstream.Subscribe(0x7F).OnObserve(convert)

    return streamA.
        ZipFromIterable(streamB, zipper).
        ZipFromIterable(streamC, zipper).
        ZipFromIterable(streamD, zipper).
        ZipFromIterable(streamE, zipper).
        StdOut().
	Encode(0x11)
}

The problem, is that only the first two packets sent by the source are received on the flow. The order of zipping doesn't matter, the order of sending by the source matters only, for what I can see...

What can I do? (Also, I'm zipping sequentially, because there's no Merge operator... Is it really missing?)

Originally posted by @kuredoro in #158