public void run()

in local/twitter-userstream-elasticsearch/src/main/java/org/apache/streams/example/TwitterUserstreamElasticsearch.java [87:114]
22 lines of code
1 McCabe index (conditional complexity)

  public void run() {

    TwitterStreamConfiguration twitterStreamConfiguration = config.getTwitter();
    ElasticsearchWriterConfiguration elasticsearchWriterConfiguration = config.getElasticsearch();

    TwitterStreamProvider stream = new TwitterStreamProvider(twitterStreamConfiguration);
    ActivityConverterProcessor converter = new ActivityConverterProcessor();
    VerbDefinitionDropFilter noDeletesProcessor = new VerbDefinitionDropFilter(Stream.of(deleteVerbDefinition).collect(Collectors.toSet()));
    ElasticsearchPersistWriter writer = new ElasticsearchPersistWriter(elasticsearchWriterConfiguration);
    VerbDefinitionKeepFilter deleteOnlyProcessor = new VerbDefinitionKeepFilter(Stream.of(deleteVerbDefinition).collect(Collectors.toSet()));
    SetDeleteIdProcessor setDeleteIdProcessor = new SetDeleteIdProcessor();
    ElasticsearchPersistDeleter deleter = new ElasticsearchPersistDeleter(elasticsearchWriterConfiguration);

    LocalRuntimeConfiguration localRuntimeConfiguration =
        StreamsJacksonMapper.getInstance().convertValue(StreamsConfigurator.detectConfiguration(), LocalRuntimeConfiguration.class);
    StreamBuilder builder = new LocalStreamBuilder(localRuntimeConfiguration);

    builder.newPerpetualStream(TwitterStreamProvider.class.getCanonicalName(), stream);
    builder.addStreamsProcessor(ActivityConverterProcessor.class.getCanonicalName(), converter, 2, TwitterStreamProvider.class.getCanonicalName());
    builder.addStreamsProcessor(VerbDefinitionDropFilter.class.getCanonicalName(), noDeletesProcessor, 1, ActivityConverterProcessor.class.getCanonicalName());
    builder.addStreamsPersistWriter(ElasticsearchPersistWriter.class.getCanonicalName(), writer, 1, VerbDefinitionDropFilter.class.getCanonicalName());
    builder.addStreamsProcessor(VerbDefinitionKeepFilter.class.getCanonicalName(), deleteOnlyProcessor, 1, ActivityConverterProcessor.class.getCanonicalName());
    builder.addStreamsProcessor(SetDeleteIdProcessor.class.getCanonicalName(), setDeleteIdProcessor, 1, VerbDefinitionKeepFilter.class.getCanonicalName());
    builder.addStreamsPersistWriter(ElasticsearchPersistDeleter.class.getCanonicalName(), deleter, 1, SetDeleteIdProcessor.class.getCanonicalName());

    builder.start();

  }