add GetEventWithStartTime (go-mysql-org#327)

* add GetEventWithStartTime

replication/binlogstreamer.go

@@ -2,7 +2,7 @@ package replication
 
 import (
 	"context"
-
+	"time"
 	"github.com/juju/errors"
 	"github.com/siddontang/go-log/log"
 )
@@ -36,6 +36,26 @@ func (s *BinlogStreamer) GetEvent(ctx context.Context) (*BinlogEvent, error) {
 	}
 }
 
+// Get the binlog event with starttime, if current binlog event timestamp smaller than specify starttime
+// return nil event 
+func (s *BinlogStreamer) GetEventWithStartTime(ctx context.Context,startTime time.Time) (*BinlogEvent, error) {
+	if s.err != nil {
+		return nil, ErrNeedSyncAgain
+	}
+	startUnix := startTime.Unix()
+	select {
+	case c := <-s.ch:
+		if int64(c.Header.Timestamp) >= startUnix {
+			return c, nil
+		}
+		return nil,nil
+	case s.err = <-s.ech:
+		return nil, s.err
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
 // DumpEvents dumps all left events
 func (s *BinlogStreamer) DumpEvents() []*BinlogEvent {
 	count := len(s.ch)