title: Write data from custom source to flink in continuous way
body: <p>It is the first time I am working with Apache Flink (1.3.1) and have a question. In more detail I am working with flink-core, flink-cep and flink-streaming library. My application is an Akka ActorSystem what consumes messages from RabbitMQ and various actors handle this messages. In some actors I want to instantiate a <code>StreamExecutionEnvironment</code> from Flink and process the incoming messages. Therefore I wrote a custom source class what extends the <code>RichSourceFunction</code> class. Everything works find, except one thing: I do not know how to send data to my Flink extension. Here is my setup:</p>
  
  <pre><code>public class FlinkExtension {
  
      private static StreamExecutionEnvironment environment;
      private DataStream&lt;ValueEvent&gt; input;
      private CustomSourceFunction function;
  
      public FlinkExtension(){
  
          environment = StreamExecutionEnvironment.getExecutionEnvironment();
  
          function = new CustomSourceFunction();
          input = environment.addSource(function);
  
          PatternStream&lt;ValueEvent&gt; patternStream = CEP.pattern(input, _pattern());
  
          DataStream&lt;String&gt; warnings = patternStream.select(new PatternSelectFunction&lt;ValueEvent, String&gt;() {
              @Override
              public String select(Map&lt;String, List&lt;ValueEvent&gt;&gt; pattern) throws Exception {
                  return null; //TODO
              }
          });
  
          warnings.print();
  
          try {
              environment.execute();
          } catch(Exception e){
              e.printStackTrace();
          }
  
      }
  
      private Pattern&lt;ValueEvent, ?&gt; _pattern(){
  
          return Pattern.&lt;ValueEvent&gt;begin("first").where(new SimpleCondition&lt;ValueEvent&gt;() {
              @Override
              public boolean filter(ValueEvent value) throws Exception {
                  return value.getValue() &gt; 10;
              }
          });
      }
  
      public void sendData(ValueEvent value){
          function.sendData(value);
      }
  }
  </code></pre>
  
  <p>And this is my custom source function:</p>
  
  <pre><code>public class CustomSourceFunction extends RichSourceFunction&lt;ValueEvent&gt; {
  
      private volatile boolean run = false;
      private SourceContext&lt;ValueEvent&gt; context;
  
      @Override
      public void open(Configuration parameters){
          run = true;
      }
  
      @Override
      public void run(SourceContext&lt;ValueEvent&gt; ctx) throws Exception {
          this.context = ctx;
  
          while (run){
  
          }
      }
  
      public void sendData(ValueEvent value){
          this.context.collectWithTimestamp(value, Calendar.getInstance().getTimeInMillis());
      }
  
      @Override
      public void cancel() {
          run = false;
      }
  }
  </code></pre>
  
  <p>So I want to call the method <code>sendData</code> in my <code>FlinkExtension</code> class from outside to write data in a continuous way to my <code>FlinkExtension</code>. Here is my JUnit test what should send data to the extension and then write the data to the <code>SourceContext</code>. </p>
  
  <pre><code>@Test
  public void testSendData(){
      FlinkExtension extension = new FlinkExtension();
      extension.sendData(new ValueEvent(30));
  }
  </code></pre>
  
  <p>But if I run the test, nothing happens, the application hangs in the run method of the <code>CustomSourceFunction</code>. I also tried to create a new endless thread in the <code>CustomSourceFunction</code> run method. </p>
  
  <p>To summarize: Does anybody know how to write data from an application to a Flink instance in a continuous way?</p>