Merge pull request #44 from lovoo/recovered

methods for checking recovery status

processor.go

@@ -561,7 +561,7 @@ func (g *Processor) createPartition(id int32) error {
 		}
 	}
 	for _, v := range g.views {
-		wait = append(wait, v.Ready)
+		wait = append(wait, v.Recovered)
 	}
 
 	g.partitions[id] = newPartition(
@@ -710,3 +710,28 @@ func (g *Processor) process(msg *message, st storage.Storage, wg *sync.WaitGroup
 
 	return nil
 }
+
+// Recovered returns true when the processor has caught up with events from kafka.
+func (g *Processor) Recovered() bool {
+	for _, v := range g.views {
+		if !v.Recovered() {
+			return false
+		}
+	}
+
+	for _, part := range g.partitionViews {
+		for _, topicPart := range part {
+			if !topicPart.ready() {
+				return false
+			}
+		}
+	}
+
+	for _, p := range g.partitions {
+		if !p.ready() {
+			return false
+		}
+	}
+
+	return true
+}

view.go

@@ -302,8 +302,8 @@ func (v *View) run() {
 	}
 }
 
-// Ready returns true when the view has caught up with events from kafka.
-func (v *View) Ready() bool {
+// Recovered returns true when the view has caught up with events from kafka.
+func (v *View) Recovered() bool {
 	for _, p := range v.partitions {
 		if !p.ready() {
 			return false