func (s *server) InitSlave(ctx context.Context, request *pb.InitSlaveRequest) (*pb.InitSlaveResponse, error) {
	ctx = callinfo.GRPCCallInfo(ctx)
	response := &pb.InitSlaveResponse{}
	return response, s.agent.RPCWrapLockAction(ctx, actionnode.TabletActionInitSlave, request, response, true, func() error {
		return s.agent.InitSlave(ctx, topo.ProtoToTabletAlias(request.Parent), myproto.ProtoToReplicationPosition(request.ReplicationPosition), request.TimeCreatedNs)
	})
}

// StreamUpdate is part of the pbs.UpdateStreamServer interface
func (server *UpdateStream) StreamUpdate(req *pb.StreamUpdateRequest, stream pbs.UpdateStream_StreamUpdateServer) (err error) {
	defer server.updateStream.HandlePanic(&err)
	return server.updateStream.ServeUpdateStream(&proto.UpdateStreamRequest{
		Position: myproto.ProtoToReplicationPosition(req.Position),
	}, func(reply *proto.StreamEvent) error {
		return stream.Send(&pb.StreamUpdateResponse{
			StreamEvent: proto.StreamEventToProto(reply),
		})
	})
}

func (s *server) PromoteSlaveWhenCaughtUp(ctx context.Context, request *pb.PromoteSlaveWhenCaughtUpRequest) (*pb.PromoteSlaveWhenCaughtUpResponse, error) {
	ctx = callinfo.GRPCCallInfo(ctx)
	response := &pb.PromoteSlaveWhenCaughtUpResponse{}
	return response, s.agent.RPCWrapLockAction(ctx, actionnode.TabletActionPromoteSlaveWhenCaughtUp, request, response, true, func() error {
		position, err := s.agent.PromoteSlaveWhenCaughtUp(ctx, myproto.ProtoToReplicationPosition(request.Position))
		if err == nil {
			response.Position = myproto.ReplicationPositionToProto(position)
		}
		return err
	})
}

func (s *server) StopSlaveMinimum(ctx context.Context, request *pb.StopSlaveMinimumRequest) (*pb.StopSlaveMinimumResponse, error) {
	ctx = callinfo.GRPCCallInfo(ctx)
	response := &pb.StopSlaveMinimumResponse{}
	return response, s.agent.RPCWrapLock(ctx, actionnode.TabletActionStopSlaveMinimum, request, response, true, func() error {
		position, err := s.agent.StopSlaveMinimum(ctx, myproto.ProtoToReplicationPosition(request.Position), time.Duration(request.WaitTimeout))
		if err == nil {
			response.Position = myproto.ReplicationPositionToProto(position)
		}
		return err
	})
}

// PromoteSlave is part of the tmclient.TabletManagerClient interface
func (client *Client) PromoteSlave(ctx context.Context, tablet *topo.TabletInfo) (myproto.ReplicationPosition, error) {
	cc, c, err := client.dial(ctx, tablet)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	defer cc.Close()
	response, err := c.PromoteSlave(ctx, &pb.PromoteSlaveRequest{})
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	return myproto.ProtoToReplicationPosition(response.Position), err
}

// StreamTables is part of the pbs.UpdateStreamServer interface
func (server *UpdateStream) StreamTables(req *pb.StreamTablesRequest, stream pbs.UpdateStream_StreamTablesServer) (err error) {
	defer server.updateStream.HandlePanic(&err)
	return server.updateStream.StreamTables(&proto.TablesRequest{
		Position: myproto.ProtoToReplicationPosition(req.Position),
		Tables:   req.Tables,
		Charset:  mproto.ProtoToCharset(req.Charset),
	}, func(reply *proto.BinlogTransaction) error {
		return stream.Send(&pb.StreamTablesResponse{
			BinlogTransaction: proto.BinlogTransactionToProto(reply),
		})
	})
}

// StreamKeyRange is part of the pbs.UpdateStreamServer interface
func (server *UpdateStream) StreamKeyRange(req *pb.StreamKeyRangeRequest, stream pbs.UpdateStream_StreamKeyRangeServer) (err error) {
	defer server.updateStream.HandlePanic(&err)
	return server.updateStream.StreamKeyRange(&proto.KeyRangeRequest{
		Position:       myproto.ProtoToReplicationPosition(req.Position),
		KeyspaceIdType: key.ProtoToKeyspaceIdType(req.KeyspaceIdType),
		KeyRange:       key.ProtoToKeyRange(req.KeyRange),
		Charset:        mproto.ProtoToCharset(req.Charset),
	}, func(reply *proto.BinlogTransaction) error {
		return stream.Send(&pb.StreamKeyRangeResponse{
			BinlogTransaction: proto.BinlogTransactionToProto(reply),
		})
	})
}

// RunBlpUntil is part of the tmclient.TabletManagerClient interface
func (client *Client) RunBlpUntil(ctx context.Context, tablet *topo.TabletInfo, positions *blproto.BlpPositionList, waitTime time.Duration) (myproto.ReplicationPosition, error) {
	cc, c, err := client.dial(ctx, tablet)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	defer cc.Close()
	response, err := c.RunBlpUntil(ctx, &pb.RunBlpUntilRequest{
		BlpPositions: blproto.BlpPositionListToProto(positions),
		WaitTimeout:  int64(waitTime),
	})
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	return myproto.ProtoToReplicationPosition(response.Position), nil
}

// StopSlaveMinimum is part of the tmclient.TabletManagerClient interface
func (client *Client) StopSlaveMinimum(ctx context.Context, tablet *topo.TabletInfo, minPos myproto.ReplicationPosition, waitTime time.Duration) (myproto.ReplicationPosition, error) {
	cc, c, err := client.dial(ctx, tablet)
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	defer cc.Close()
	response, err := c.StopSlaveMinimum(ctx, &pb.StopSlaveMinimumRequest{
		Position:    myproto.ReplicationPositionToProto(minPos),
		WaitTimeout: int64(waitTime),
	})
	if err != nil {
		return myproto.ReplicationPosition{}, err
	}
	return myproto.ProtoToReplicationPosition(response.Position), err
}

// ProtoToBlpPosition converts a proto to a BlpPosition
func ProtoToBlpPosition(b *pbt.BlpPosition) *BlpPosition {
	return &BlpPosition{
		Uid:      b.Uid,
		Position: myproto.ProtoToReplicationPosition(b.Position),
	}
}

func (s *server) PopulateReparentJournal(ctx context.Context, request *pb.PopulateReparentJournalRequest) (*pb.PopulateReparentJournalResponse, error) {
	ctx = callinfo.GRPCCallInfo(ctx)
	response := &pb.PopulateReparentJournalResponse{}
	return response, s.agent.RPCWrap(ctx, actionnode.TabletActionPopulateReparentJournal, request, response, func() error {
		return s.agent.PopulateReparentJournal(ctx, request.TimeCreatedNs, request.ActionName, topo.ProtoToTabletAlias(request.MasterAlias), myproto.ProtoToReplicationPosition(request.ReplicationPosition))
	})
}