How zip 3 or more streams with different keys?

[kuredoro]

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?)

[xiaojian-hong]

Hi @kuredoro ,
Thanks for your feedback.
YoMo supposes to zip 3 or more streams, we will fix this issue.

[kuredoro]

Okay....

Also, as a side question: what y3 keys should be working? I tried different keys: got panic at 0x00 - 0x09 and 0x80-0xFF, but then also got that the keys in range 0x40 - 0x7F just cannot be received :(

[xiaojian-hong]

Y3 preserves the keys 0x00 - 0x0f for internal use, we expect 0x10 - 0x7f should be working.
Seems there is an issue to decode the keys 0x40 - 0x7f, we will fix it as well.

[xiaojian-hong]

Hi @kuredoro ,
Sorry, I just got the confirmation from the maintainer @chenjunbiao of our Y3 Codec project. Unfortunately, Y3 only supports the keys in the range 0x10 - 0x3f.
According to the design of the Y3 version draft-01, the least significant 6 bits of the Tag byte are using for the key, which means the range is 0x00 - 0x7f.
We reserved the range 0x00 - 0x0f for internal use, so only 0x10 - 0x3f left for user-defined keys.

We will expand the range of Y3 keys in the version draft-02.

[kuredoro]

That's fine.

[fanweixiao]

thanks @kuredoro , add a bug issue: #159, lets solve it ;)

[fanweixiao]

solved by #162

[fanweixiao]

@kuredoro please see #162 , version v1.1.0 solve the problem, please check 💯

[kuredoro]

Hi! Stream zipping works great on my behalf. Thanks for solving!

Although, I had a strange thing happen with the code I posted in the beginning of this thread: the last packet sent from the source, for some reason was ignored by the flow, so it could not print out the complete slice of 5 elements. Why could this be, any ideas?

If I send the last packet it twice, it prints the slice and then panics in Encode method... because yomo can't encode slices, I presume, in which case it's fine...

[fanweixiao]

YoMo didn't process the codec issue, right now, YoMo only accepts the y3-codec format, and it support slice. btw, in the next version 2~3 weeks later, YoMo will support JSON :)

@chenjunbiao can you share examples on how to deal with slice in y3-codec?

[chenjunbiao]

Thanks for your attention! Examples of using slice：

• slice1: for base types
• slice2: for struct
• types description