How do I split and merge a Multi into a variable number of Multis?

[porcelijn]

I'm building a stream processor that ingests and generates a multiplexed stream. (For those familiar with media stream, think: MPEG Transport Stream, linear satellite broadcast, but it equally applies to group chat sessions.)

I need to demultiplex the stream and process each individual channel separately before muxing them back again, so something like Multi.split(keyType, splitter) followed by Multi.createBy().merging().streams(...) looks like a natural fit.

The problem in my case is that the pipeline structure is dynamic. Embedded streams can come and go more or less asynchronously and the stream (ingest & egress) should be running before it's configuration (or even the number of sub-streams) is known.

So far, I've just done this all manually, like so:

        class Dispatcher {
            private Multi<String> incoming;
            private Map<Long, Multi<String>> channels = new HashMap<>();
            UnicastProcessor<String> outgoing = UnicastProcessor.create();

            Dispatcher(Multi<String> incoming) {
                this.incoming = incoming.broadcast()
                        .withCancellationAfterLastSubscriberDeparture()
                        .toAtLeast(1);
                incoming.subscribe().with(n -> {
                    if (!ids.keySet().stream().anyMatch(id -> (n % id == 0))) {
                        System.err.printf("Channel %d not registered in %s\n", n, ids.keySet());
                    }
                });
            }

            void registerStream(long id, Function<Multi<String>, Multi<String>> handler) {
                ids.put(id, incoming.filter(m -> getChannelId(m) == id));
                ids.get(id).subscribe().with(m -> {
                    outgoing.onNext(handler.apply(getPayload(m));
                });
            }

            Multi<String> merged() {
                return outgoing;
            }
            
            private static long getChannelId(String message) {...}
            private static String getPayload(String message) {...}
        }

       static void main() {
           Multi<String> stream = ...
           var d = new Dispatcher(stream);
           d.merged().subscribe().with(System.out::println);
           //...
           d.register(1, m -> m.toUpperCase());
           // ...
           d.register(2, m -> m.toLowerCase());   
       }

...but I was wondering if there is a more idiomatic way to approach this.
Any suggestions?

[jponge]

I don't have any good recommendation now.

The original splitting feature was motivated by needs in Reactive Messaging (/cc @ozangunalp) and it was a sound assumption that you'd split into a finite amount of streams based on a key set.

You might want to check https://javadoc.io/doc/io.smallrye.reactive/mutiny/latest/io.smallrye.mutiny/io/smallrye/mutiny/groups/MultiGroup.html groupBy, but that might not solve the "merge back" piece.

[jponge]

Actually something you could do to merge back a dynamic amount of streams is to:

1. define a Multi emitter / UnicastProcessor (see https://smallrye.io/smallrye-mutiny/latest/guides/integrate-a-non-reactive-source/)
2. have some dynamic split
3. subscribe to each sub-stream
4. for each item on each sub-stream, push to the processor / emitter

That's not perfect and possibly dirty, but... it might be it

[porcelijn]

Wicked! Thanks so much, that is very helpful!
...So you're suggesting something like this?

    record Message(long streamId, String payload) {
        static Message alterPayload(Message message) {
            return new Message(message.streamId, message.payload.toUpperCase());
        }
    }
     
    public void main() {
        Map<Long, Multi<Message>> subStreams = new ConcurrentHashMap<>();
        {
            Multi<Message> stream =
                    Multi.createFrom().ticks().every(Duration.ofMillis(10)).map(n -> new Message(n % 5, "foo" + n));
            Multi<GroupedMulti<Long, Message>> groupedStreams = stream.group().by(Message::streamId);
            groupedStreams
                    .subscribe()
                    .with(
                            group -> {
                                long id = group.key();
                                boolean streamOfInterest = id < 2;
                                if (streamOfInterest) {
                                    System.out.printf("Saving stream with id=%d for future access\n", id);
                                    subStreams.put(id, group);
                                } else {
                                    System.out.printf("Ignoring stream with id=%d\n", id);
                                }
                            },
                            failure -> {
                                throw new RuntimeException(failure);
                            });
        }
        // ...And then later
        Thread.sleep(100);
        UnicastProcessor<String> merged = UnicastProcessor.create();
        {
            long audio = 1;
            System.out.printf("Subscribing to audio stream id=%d\n", audio);
            subStreams
                    .get(audio)
                    .map(Message::alterPayload)
                    .subscribe()
                    .with(
                            m -> {
                                merged.onNext(String.format("[Group with id=%d] Received audio: %s", audio, m));
                            },
                            f -> {
                                throw new RuntimeException(f);
                            });
        }
        {
            long video = 0;
            System.out.printf("Subscribing to video stream id=%d\n", video);
            subStreams
                    .get(video)
                    .map(Message::alterPayload)
                    .subscribe()
                    .with(
                            m -> {
                                merged.onNext(String.format("[Group with id=%d] Received video: %s", video, m));
                            },
                            f -> {
                                throw new RuntimeException(f);
                            });
        }
        
        merged.subscribe().with(System.out::println);
        Thread.sleep(1000);
    }

Which results in something like:

Saving stream with id=0 for future access
Saving stream with id=1 for future access
Ignoring stream with id=2
Ignoring stream with id=3
Ignoring stream with id=4
Subscribing to audio stream id=1
Subscribing to video stream id=0
[Group with id=1] Received audio: Message[streamId=1, payload=FOO1]
[Group with id=1] Received audio: Message[streamId=1, payload=FOO6]
[Group with id=1] Received audio: Message[streamId=1, payload=FOO11]
[Group with id=0] Received video: Message[streamId=0, payload=FOO0]
[Group with id=0] Received video: Message[streamId=0, payload=FOO5]
[Group with id=0] Received video: Message[streamId=0, payload=FOO10]
[Group with id=0] Received video: Message[streamId=0, payload=FOO15]
[Group with id=1] Received audio: Message[streamId=1, payload=FOO16]
[Group with id=0] Received video: Message[streamId=0, payload=FOO20]
[Group with id=1] Received audio: Message[streamId=1, payload=FOO21]
[Group with id=0] Received video: Message[streamId=0, payload=FOO25]
[Group with id=1] Received audio: Message[streamId=1, payload=FOO26]
[Group with id=0] Received video: Message[streamId=0, payload=FOO30]
[Group with id=1] Received audio: Message[streamId=1, payload=FOO31]
...

[jponge]

If it's doing what you need I guess so 😇

[porcelijn]

😄 🤔 It wasn't, but here is a better version:

    public void main() {
        record Message(long streamId, String payload) {
            static Message alterAudio(Message message) {
                return new Message(message.streamId, message.payload.toUpperCase());
            }
            static Message alterVideo(Message message) {
                return new Message(message.streamId, message.payload.toLowerCase());
            }
        }

        class Dispatcher {
            Map<Long, UnicastProcessor<Message>> subStreams = new ConcurrentHashMap<>();
            BroadcastProcessor<String> merged = BroadcastProcessor.create();

            void subscribeTo(long streamId, Function<Message, Message> mapper) {
                System.out.printf("Subscribing to stream id=%d, %s\n", streamId,
                subStreams.putIfAbsent(streamId, UnicastProcessor.create()) == null ? "absent" : "present");
                subStreams
                        .get(streamId)
                        .map(mapper)
                        .subscribe()
                        .with(
                                m -> {
                                    merged.onNext(String.format("[Group with id=%d] Received: %s", streamId, m));
                                },
                                merged::onError,
                                merged::onComplete);
            }

            void attachTo(Multi<Message> stream) {
                Multi<GroupedMulti<Long, Message>> groupedStreams =
                        stream.group().by(Message::streamId);
                groupedStreams
                        .subscribe()
                        .with(
                                group -> {
                                    long id = group.key();
                                    boolean streamOfInterest = id < 3;
                                    if (streamOfInterest) {
                                        System.out.printf("Saving stream with id=%d for %s use\n", id,
                                        subStreams.putIfAbsent(id, UnicastProcessor.create()) ==null ? "future":"immediate");
                                        UnicastProcessor<Message> subStream = subStreams.get(id);
                                        group.subscribe()
                                                .with(
                                                        m -> {
                                                            if (subStream.hasSubscriber()) {
                                                                subStream.onNext(m);
                                                            }
                                                        },
                                                        f -> {
                                                            throw new RuntimeException(f);
                                                        });
                                    } else {
                                        System.out.printf("Ignoring stream with id=%d\n", id);
                                    }
                                },
                                merged::onError,
                                merged::onComplete);
            }
        }

        var dispatcher = new Dispatcher();

        // Subscribe to muxed output ahead of time
        dispatcher.merged.map(s -> "Viewer 1" + s).subscribe().with(System.out::println);

        // Subscribe to video stream ahead of time
        dispatcher.subscribeTo(0, Message::alterVideo);

        Multi<Message> stream = Multi.createFrom()
                .ticks()
                .every(Duration.ofMillis(10))
                .map(n -> new Message(n % 5, "Foo" + n))
                .select()
                .first(67);

        Thread.sleep(100);

        // ... Now start ingest
        dispatcher.attachTo(stream);

        Thread.sleep(100);

        // ... Subscribe to audio while streaming in progress
        dispatcher.subscribeTo(1, Message::alterAudio);

        // ... Tune in to muxed output
        dispatcher.merged.map(s -> "Viewer 2" + s).subscribe().with(System.out::println);

        Thread.sleep(1000);
    }

Gets you something like:

Subscribing to stream id=0, absent
Saving stream with id=0 for immediate use
Viewer 1[Group with id=0] Received: Message[streamId=0, payload=foo0]
Saving stream with id=1 for future use
Saving stream with id=2 for future use
Ignoring stream with id=3
Ignoring stream with id=4
Viewer 1[Group with id=0] Received: Message[streamId=0, payload=foo5]
Viewer 1[Group with id=0] Received: Message[streamId=0, payload=foo10]
Subscribing to stream id=1, present
Viewer 1[Group with id=1] Received: Message[streamId=1, payload=FOO11]
Viewer 2[Group with id=1] Received: Message[streamId=1, payload=FOO11]
Viewer 1[Group with id=0] Received: Message[streamId=0, payload=foo15]
Viewer 2[Group with id=0] Received: Message[streamId=0, payload=foo15]
...

So indeed, these UnicastProcessor / BroadcastProcessor objects are very versatile and much more convenient than Multi.emitter(). Specifically, the example shows there is no temporal coupling — it's possible to register sub-stream subscriptions ahead of time or later once the ingest has started.

@jponge Thanks for helping out!