@Override
    public StateFactory<State> init(SourceEventListener sourceEventListener, OptionHolder optionHolder,
                                    String[] requestedTransportPropertyNames, ConfigReader configReader,
                                    SiddhiAppContext siddhiAppContext) {
        super.init(sourceEventListener, optionHolder, requestedTransportPropertyNames, configReader, siddhiAppContext);
        symbol = optionHolder.validateAndGetStaticValue("prop1");
        price = optionHolder.validateAndGetStaticValue("prop2");
        fail = optionHolder.validateAndGetStaticValue("fail", "false");
        this.sourceEventListener = sourceEventListener;
        String topic = optionHolder.validateAndGetStaticValue(TOPIC_KEY, "input inMemory source");
        this.subscriber = new InMemoryBroker.Subscriber() {
            @Override
            public void onMessage(Object event) {
                if (fail.equals("true")) {
                    sourceEventListener.onEvent(event, new String[]{symbol});
                } else {
                    if (requestedTransportPropertyNames.length == 2 &&
                            requestedTransportPropertyNames[0].equals("symbol") &&
                            requestedTransportPropertyNames[1].equals("price")) {
                        sourceEventListener.onEvent(event, new String[]{symbol, price});
                    }
                }
            }

            @Override
            public String getTopic() {
                return topic;
            }
        };
        return null;
    }
Implementation of {@link Source} to receive events through in-memory
transport.
modules/siddhi-core/src/test/java/io/siddhi/core/transport/TestTrpInMemorySource.java