Merge branch 'replication'

go/vt/binlog/binlogplayer/client.go

@@ -30,6 +30,9 @@ type BinlogPlayerClient interface {
 	// Close the connection
 	Close()
 
+	// Ask the server to stream binlog updates
+	ServeUpdateStream(*proto.UpdateStreamRequest, chan *proto.StreamEvent) BinlogPlayerResponse
+
 	// Ask the server to stream updates related to the provided tables
 	StreamTables(*proto.TablesRequest, chan *proto.BinlogTransaction) BinlogPlayerResponse
 

go/vt/binlog/gorpcbinlogplayer/player.go

@@ -34,8 +34,8 @@ func (client *GoRpcBinlogPlayerClient) Close() {
 	client.Client.Close()
 }
 
-func (client *GoRpcBinlogPlayerClient) StreamTables(req *proto.TablesRequest, responseChan chan *proto.BinlogTransaction) binlogplayer.BinlogPlayerResponse {
-	resp := client.Client.StreamGo("UpdateStream.StreamTables", req, responseChan)
+func (client *GoRpcBinlogPlayerClient) ServeUpdateStream(req *proto.UpdateStreamRequest, responseChan chan *proto.StreamEvent) binlogplayer.BinlogPlayerResponse {
+	resp := client.Client.StreamGo("UpdateStream.ServeUpdateStream", req, responseChan)
 	return &GoRpcBinlogPlayerResponse{resp}
 }
 
@@ -44,6 +44,11 @@ func (client *GoRpcBinlogPlayerClient) StreamKeyRange(req *proto.KeyRangeRequest
 	return &GoRpcBinlogPlayerResponse{resp}
 }
 
+func (client *GoRpcBinlogPlayerClient) StreamTables(req *proto.TablesRequest, responseChan chan *proto.BinlogTransaction) binlogplayer.BinlogPlayerResponse {
+	resp := client.Client.StreamGo("UpdateStream.StreamTables", req, responseChan)
+	return &GoRpcBinlogPlayerResponse{resp}
+}
+
 // Registration as a factory
 func init() {
 	binlogplayer.RegisterBinlogPlayerClientFactory("gorpc", func() binlogplayer.BinlogPlayerClient {

test/environment.py

@@ -126,3 +126,5 @@ def tablet_manager_protocol_flags():
 def vtgate_protocol_flags():
   return ['-tablet-protocol', 'bson']
 
+def binlog_player_protocol_flags():
+  return ['-binlog_player_protocol', 'gorpc']

test/tablet.py

@@ -287,6 +287,7 @@ class Tablet(object):
             '-tablet-path', self.tablet_alias,
             '-log_dir', environment.vtlogroot]
     args.extend(environment.topo_server_flags())
+    args.extend(environment.binlog_player_protocol_flags())
 
     dbconfigs = self._get_db_configs_file(repl_extra_flags)
     for key1 in dbconfigs: