MessageStreamImpl implements MessageStream<M> { @Override public MessageStream<M> merge(Collection<? extends MessageStream<? extends M>> otherStreams) { if (otherStreams.isEmpty()) return this; String opId = this.streamAppDesc.getNextOpId(OpCode.MERGE); StreamOperatorSpec<M, M> op = OperatorSpecs.createMergeOperatorSpec(opId); this.operatorSpec.registerNextOperatorSpec(op); otherStreams.forEach(other -> ((MessageStreamImpl<M>) other).getOperatorSpec().registerNextOperatorSpec(op)); return new MessageStreamImpl<>(this.streamAppDesc, op); } MessageStreamImpl(StreamApplicationDescriptorImpl streamAppDesc, OperatorSpec<?, M> operatorSpec); @Override MessageStream<TM> map(MapFunction<? super M, ? extends TM> mapFn); @Override MessageStream<M> filter(FilterFunction<? super M> filterFn); @Override MessageStream<TM> flatMap(FlatMapFunction<? super M, ? extends TM> flatMapFn); @Override MessageStream<OM> flatMapAsync(AsyncFlatMapFunction<? super M, ? extends OM> flatMapFn); @Override void sink(SinkFunction<? super M> sinkFn); @Override MessageStream<M> sendTo(OutputStream<M> outputStream); @Override MessageStream<WindowPane<K, WV>> window(Window<M, K, WV> window, String userDefinedId); @Override MessageStream<JM> join(MessageStream<OM> otherStream,
      JoinFunction<? extends K, ? super M, ? super OM, ? extends JM> joinFn,
      Serde<K> keySerde, Serde<M> messageSerde, Serde<OM> otherMessageSerde,
      Duration ttl, String userDefinedId); @Override MessageStream<JM> join(Table<R> table,
      StreamTableJoinFunction<? extends K, ? super M, ? super R, ? extends JM> joinFn, Object ... args); @Override MessageStream<M> merge(Collection<? extends MessageStream<? extends M>> otherStreams); @Override MessageStream<KV<K, V>> partitionBy(MapFunction<? super M, ? extends K> keyExtractor,
      MapFunction<? super M, ? extends V> valueExtractor, KVSerde<K, V> serde, String userDefinedId); @Override MessageStream<KV<K, V>> sendTo(Table<KV<K, V>> table, Object ... args); @Override MessageStream<M> broadcast(Serde<M> serde, String userDefinedId); @VisibleForTesting OperatorSpec<?, M> getOperatorSpec();  }
@Test public void testMerge() { StreamApplicationDescriptorImpl mockGraph = mock(StreamApplicationDescriptorImpl.class); OperatorSpec mockOpSpec1 = mock(OperatorSpec.class); MessageStream<TestMessageEnvelope> inputStream = new MessageStreamImpl<>(mockGraph, mockOpSpec1); OperatorSpec mockOpSpec2 = mock(OperatorSpec.class); OperatorSpec mockOpSpec3 = mock(OperatorSpec.class); Collection<MessageStream<TestMessageEnvelope>> otherStreams1 = ImmutableList.of( new MessageStreamImpl<>(mockGraph, mockOpSpec2), new MessageStreamImpl<>(mockGraph, mockOpSpec3) ); inputStream.merge(otherStreams1); ArgumentCaptor<OperatorSpec> registeredOpCaptor1 = ArgumentCaptor.forClass(OperatorSpec.class); verify(mockOpSpec1).registerNextOperatorSpec(registeredOpCaptor1.capture()); OperatorSpec<?, TestMessageEnvelope> registeredOpSpec1 = registeredOpCaptor1.getValue(); assertTrue(registeredOpSpec1 instanceof StreamOperatorSpec); FlatMapFunction transformFn = ((StreamOperatorSpec) registeredOpSpec1).getTransformFn(); ArgumentCaptor<OperatorSpec> registeredOpCaptor2 = ArgumentCaptor.forClass(OperatorSpec.class); verify(mockOpSpec2).registerNextOperatorSpec(registeredOpCaptor2.capture()); OperatorSpec<?, TestMessageEnvelope> registeredOpSpec2 = registeredOpCaptor2.getValue(); ArgumentCaptor<OperatorSpec> registeredOpCaptor3 = ArgumentCaptor.forClass(OperatorSpec.class); verify(mockOpSpec3).registerNextOperatorSpec(registeredOpCaptor3.capture()); OperatorSpec<?, TestMessageEnvelope> registeredOpSpec3 = registeredOpCaptor3.getValue(); assertEquals(registeredOpSpec1, registeredOpSpec2); assertEquals(registeredOpSpec2, registeredOpSpec3); assertEquals(OpCode.MERGE, registeredOpSpec1.getOpCode()); assertNotNull(transformFn); TestMessageEnvelope mockInput = mock(TestMessageEnvelope.class); assertTrue(transformFn.apply(mockInput).contains(mockInput)); assertEquals(1, transformFn.apply(mockInput).size()); }
@Test public void testMergeWithRelaxedTypes() { StreamApplicationDescriptorImpl mockGraph = mock(StreamApplicationDescriptorImpl.class); MessageStream<TestMessageEnvelope> inputStream = new MessageStreamImpl<>(mockGraph, mock(OperatorSpec.class)); Collection<MessageStream<TestMessageEnvelope>> otherStreams1 = ImmutableList.of( new MessageStreamImpl<>(mockGraph, mock(OperatorSpec.class)), new MessageStreamImpl<>(mockGraph, mock(OperatorSpec.class)) ); Collection<MessageStream<TestInputMessageEnvelope>> otherStreams2 = ImmutableList.of( new MessageStreamImpl<TestInputMessageEnvelope>(mockGraph, mock(OperatorSpec.class)), new MessageStreamImpl<TestInputMessageEnvelope>(mockGraph, mock(OperatorSpec.class)) ); Collection<MessageStream<TestMessageEnvelope>> otherStreams3 = ImmutableList.of( new MessageStreamImpl<TestMessageEnvelope>(mockGraph, mock(OperatorSpec.class)), (MessageStream) new MessageStreamImpl<TestInputMessageEnvelope>(mockGraph, mock(OperatorSpec.class)) ); Collection<MessageStream<? extends TestMessageEnvelope>> otherStreams4 = ImmutableList.of( new MessageStreamImpl<TestMessageEnvelope>(mockGraph, mock(OperatorSpec.class)), new MessageStreamImpl<TestInputMessageEnvelope>(mockGraph, mock(OperatorSpec.class)) ); inputStream.merge(otherStreams1); inputStream.merge(otherStreams2); inputStream.merge(otherStreams3); inputStream.merge(otherStreams4); }
@Test public <T> void testMergeWithNestedTypes() { class MessageEnvelope<TM> { } MessageStream<MessageEnvelope<T>> ms1 = mock(MessageStreamImpl.class); MessageStream<MessageEnvelope<T>> ms2 = mock(MessageStreamImpl.class); MessageStream<MessageEnvelope<T>> ms3 = mock(MessageStreamImpl.class); Collection<MessageStream<MessageEnvelope<T>>> otherStreams = ImmutableList.of(ms2, ms3); ms1.merge(otherStreams); }
@Test public void testMergeAll() { MessageStream<TestMessageEnvelope> input1 = mock(MessageStreamImpl.class); MessageStream<TestMessageEnvelope> input2 = mock(MessageStreamImpl.class); MessageStream<TestMessageEnvelope> input3 = mock(MessageStreamImpl.class); MessageStream.mergeAll(ImmutableList.of(input1, input2, input3)); ArgumentCaptor<Collection> otherStreamsCaptor = ArgumentCaptor.forClass(Collection.class); verify(input1, times(1)).merge(otherStreamsCaptor.capture()); assertEquals(2, otherStreamsCaptor.getValue().size()); assertTrue(otherStreamsCaptor.getValue().contains(input2)); assertTrue(otherStreamsCaptor.getValue().contains(input3)); }
@Test public void testMergeAllWithRelaxedTypes() { MessageStreamImpl<TestInputMessageEnvelope> input1 = mock(MessageStreamImpl.class); MessageStreamImpl<TestMessageEnvelope> input2 = mock(MessageStreamImpl.class); Collection<MessageStream<? extends TestMessageEnvelope>> streams = ImmutableList.of(input1, input2); MessageStream.mergeAll(streams); ArgumentCaptor<Collection> otherStreamsCaptor = ArgumentCaptor.forClass(Collection.class); verify(input1, times(1)).merge(otherStreamsCaptor.capture()); assertEquals(1, otherStreamsCaptor.getValue().size()); assertTrue(otherStreamsCaptor.getValue().contains(input2)); }
@Test public <T> void testMergeAllWithNestedTypes() { class MessageEnvelope<TM> { } MessageStream<MessageEnvelope<T>> input1 = mock(MessageStreamImpl.class); MessageStream<MessageEnvelope<T>> input2 = mock(MessageStreamImpl.class); MessageStream<MessageEnvelope<T>> input3 = mock(MessageStreamImpl.class); MessageStream.mergeAll(ImmutableList.of(input1, input2, input3)); ArgumentCaptor<Collection> otherStreamsCaptor = ArgumentCaptor.forClass(Collection.class); verify(input1, times(1)).merge(otherStreamsCaptor.capture()); assertEquals(2, otherStreamsCaptor.getValue().size()); assertTrue(otherStreamsCaptor.getValue().contains(input2)); assertTrue(otherStreamsCaptor.getValue().contains(input3)); }