commit 080bdca62d56d4d798f1cf6b4b12c75d8f93bcdd
author manuzhang <owenzhang1990@gmail.com> Fri Jun 23 09:57:43 2017 +0800
committer manuzhang <owenzhang1990@gmail.com> Fri Jun 23 09:57:56 2017 +0800
tree 0ccbe3e194a6ac6c45c0f56dc1bf9cfd315a8ee2
parent 2877c81b6ffd48b7e781aa2a476728c7252e7224

[GEARPUMP-320] Handle stashed messages after onStart

Author: manuzhang <owenzhang1990@gmail.com>

Closes #192 from manuzhang/fix_start.

streaming/src/main/scala/org/apache/gearpump/streaming/task/TaskActor.scala

diff --git a/streaming/src/main/scala/org/apache/gearpump/streaming/task/TaskActor.scala b/streaming/src/main/scala/org/apache/gearpump/streaming/task/TaskActor.scala
index 1b90146..1fb61bd 100644
--- a/streaming/src/main/scala/org/apache/gearpump/streaming/task/TaskActor.scala
+++ b/streaming/src/main/scala/org/apache/gearpump/streaming/task/TaskActor.scala
@@ -275,16 +275,16 @@
 
     subscriptions.foreach(_._2.start())
 
-    stashQueue.asScala.foreach { item =>
-      handleMessages(item.sender).apply(item.msg)
-    }
-    stashQueue.clear()
-
     // Put this as the last step so that the subscription is already initialized.
     // Message sending in current Task before onStart will not be delivered to
     // target
     onStart(Instant.ofEpochMilli(startClock))
 
+    stashQueue.asScala.foreach { item =>
+      handleMessages(item.sender).apply(item.msg)
+    }
+    stashQueue.clear()
+
     taskContextData.appMaster ! GetUpstreamMinClock(taskId)
     context.become(handleMessages(sender))
   }