MessageStreamImpl implements MessageStream<M> { @Override public MessageStream<M> sendTo(OutputStream<M> outputStream) { String opId = this.streamAppDesc.getNextOpId(OpCode.SEND_TO); OutputOperatorSpec<M> op = OperatorSpecs.createSendToOperatorSpec( (OutputStreamImpl<M>) outputStream, opId); this.operatorSpec.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 testSendTo() { StreamApplicationDescriptorImpl mockGraph = mock(StreamApplicationDescriptorImpl.class); OperatorSpec mockOpSpec = mock(OperatorSpec.class); MessageStreamImpl<TestMessageEnvelope> inputStream = new MessageStreamImpl<>(mockGraph, mockOpSpec); OutputStreamImpl<TestMessageEnvelope> mockOutputStreamImpl = mock(OutputStreamImpl.class); inputStream.sendTo(mockOutputStreamImpl); ArgumentCaptor<OperatorSpec> registeredOpCaptor = ArgumentCaptor.forClass(OperatorSpec.class); verify(mockOpSpec).registerNextOperatorSpec(registeredOpCaptor.capture()); OperatorSpec<?, TestMessageEnvelope> registeredOpSpec = registeredOpCaptor.getValue(); assertTrue(registeredOpSpec instanceof OutputOperatorSpec); assertEquals(OpCode.SEND_TO, registeredOpSpec.getOpCode()); assertEquals(mockOutputStreamImpl, ((OutputOperatorSpec) registeredOpSpec).getOutputStream()); MessageStreamImpl<KV<String, TestMessageEnvelope>> keyedInputStream = new MessageStreamImpl<>(mockGraph, mockOpSpec); OutputStreamImpl<KV<String, TestMessageEnvelope>> mockKeyedOutputStreamImpl = mock(OutputStreamImpl.class); keyedInputStream.sendTo(mockKeyedOutputStreamImpl); }
@Test public void testSendToTable() { StreamApplicationDescriptorImpl mockGraph = mock(StreamApplicationDescriptorImpl.class); OperatorSpec inputOpSpec = mock(OperatorSpec.class); MessageStreamImpl<TestMessageEnvelope> source = new MessageStreamImpl<>(mockGraph, inputOpSpec); TableImpl table = new TableImpl("t1"); source.sendTo(table); ArgumentCaptor<OperatorSpec> registeredOpCaptor = ArgumentCaptor.forClass(OperatorSpec.class); verify(inputOpSpec).registerNextOperatorSpec(registeredOpCaptor.capture()); OperatorSpec<?, TestMessageEnvelope> registeredOpSpec = registeredOpCaptor.getValue(); assertTrue(registeredOpSpec instanceof SendToTableOperatorSpec); SendToTableOperatorSpec sendToTableOperatorSpec = (SendToTableOperatorSpec) registeredOpSpec; assertEquals(OpCode.SEND_TO, sendToTableOperatorSpec.getOpCode()); assertEquals(table.getTableId(), sendToTableOperatorSpec.getTableId()); }