StreamApplicationDescriptorImpl extends ApplicationDescriptorImpl<StreamApplicationDescriptor> implements StreamApplicationDescriptor { @Override public <M> MessageStream<M> getInputStream(InputDescriptor<M, ?> inputDescriptor) { SystemDescriptor systemDescriptor = inputDescriptor.getSystemDescriptor(); Optional<StreamExpander> expander = systemDescriptor.getExpander(); if (expander.isPresent()) { return expander.get().apply(this, inputDescriptor); } addInputDescriptor(inputDescriptor); String streamId = inputDescriptor.getStreamId(); Serde serde = inputDescriptor.getSerde(); KV<Serde, Serde> kvSerdes = getOrCreateStreamSerdes(streamId, serde); boolean isKeyed = serde instanceof KVSerde; InputTransformer transformer = inputDescriptor.getTransformer().orElse(null); InputOperatorSpec inputOperatorSpec = OperatorSpecs.createInputOperatorSpec(streamId, kvSerdes.getKey(), kvSerdes.getValue(), transformer, isKeyed, this.getNextOpId(OpCode.INPUT, null)); inputOperators.put(streamId, inputOperatorSpec); return new MessageStreamImpl(this, inputOperators.get(streamId)); } StreamApplicationDescriptorImpl(StreamApplication userApp, Config config); @Override MessageStream<M> getInputStream(InputDescriptor<M, ?> inputDescriptor); @Override OutputStream<M> getOutputStream(OutputDescriptor<M, ?> outputDescriptor); @Override Table<KV<K, V>> getTable(TableDescriptor<K, V, ?> tableDescriptor); @Override Set<String> getInputStreamIds(); @Override Set<String> getOutputStreamIds(); @Override Set<String> getIntermediateBroadcastStreamIds(); Map<String, InputOperatorSpec> getInputOperators(); Map<String, OutputStreamImpl> getOutputStreams(); OperatorSpecGraph getOperatorSpecGraph(); String getNextOpId(OpCode opCode, String userDefinedId); String getNextOpId(OpCode opCode); @VisibleForTesting IntermediateMessageStreamImpl<M> getIntermediateStream(String streamId, Serde<M> serde, boolean isBroadcast);  }
@Test(expected = IllegalArgumentException.class) public void testGetInputStreamWithNullSerde() { GenericSystemDescriptor sd = new GenericSystemDescriptor("mockSystem", "mockSystemFactoryClass"); GenericInputDescriptor isd = sd.getInputDescriptor("mockStreamId", null); new StreamApplicationDescriptorImpl(appDesc -> { appDesc.getInputStream(isd); }, getConfig()); }
@Test(expected = IllegalStateException.class) public void testGetSameInputStreamTwice() { String streamId = "test-stream-1"; GenericSystemDescriptor sd = new GenericSystemDescriptor("mockSystem", "mockSystemFactoryClass"); GenericInputDescriptor isd1 = sd.getInputDescriptor(streamId, mock(Serde.class)); GenericInputDescriptor isd2 = sd.getInputDescriptor(streamId, mock(Serde.class)); new StreamApplicationDescriptorImpl(appDesc -> { appDesc.getInputStream(isd1); appDesc.getInputStream(isd2); }, getConfig()); }
@Test(expected = IllegalStateException.class) public void testSetDefaultSystemDescriptorAfterGettingInputStream() { String streamId = "test-stream-1"; GenericSystemDescriptor sd = new GenericSystemDescriptor("mockSystem", "mockSystemFactoryClass"); GenericInputDescriptor isd = sd.getInputDescriptor(streamId, mock(Serde.class)); new StreamApplicationDescriptorImpl(appDesc -> { appDesc.getInputStream(isd); appDesc.withDefaultSystem(sd); }, getConfig()); }