сделал telegram_client.go

vendor/google.golang.org/grpc/internal/binarylog/binarylog.go

@@ -28,8 +28,10 @@ import (
 	"google.golang.org/grpc/internal/grpcutil"
 )
 
-// Logger is the global binary logger. It can be used to get binary logger for
-// each method.
+var grpclogLogger = grpclog.Component("binarylog")
+
+// Logger specifies MethodLoggers for method names with a Log call that
+// takes a context.
 type Logger interface {
 	GetMethodLogger(methodName string) MethodLogger
 }
@@ -37,11 +39,9 @@ type Logger interface {
 // binLogger is the global binary logger for the binary. One of this should be
 // built at init time from the configuration (environment variable or flags).
 //
-// It is used to get a methodLogger for each individual method.
+// It is used to get a MethodLogger for each individual method.
 var binLogger Logger
 
-var grpclogLogger = grpclog.Component("binarylog")
-
 // SetLogger sets the binary logger.
 //
 // Only call this at init time.
@@ -56,11 +56,11 @@ func GetLogger() Logger {
 	return binLogger
 }
 
-// GetMethodLogger returns the methodLogger for the given methodName.
+// GetMethodLogger returns the MethodLogger for the given methodName.
 //
 // methodName should be in the format of "/service/method".
 //
-// Each methodLogger returned by this method is a new instance. This is to
+// Each MethodLogger returned by this method is a new instance. This is to
 // generate sequence id within the call.
 func GetMethodLogger(methodName string) MethodLogger {
 	if binLogger == nil {
@@ -117,7 +117,7 @@ func (l *logger) setDefaultMethodLogger(ml *MethodLoggerConfig) error {
 
 // Set method logger for "service/*".
 //
-// New methodLogger with same service overrides the old one.
+// New MethodLogger with same service overrides the old one.
 func (l *logger) setServiceMethodLogger(service string, ml *MethodLoggerConfig) error {
 	if _, ok := l.config.Services[service]; ok {
 		return fmt.Errorf("conflicting service rules for service %v found", service)
@@ -131,7 +131,7 @@ func (l *logger) setServiceMethodLogger(service string, ml *MethodLoggerConfig)
 
 // Set method logger for "service/method".
 //
-// New methodLogger with same method overrides the old one.
+// New MethodLogger with same method overrides the old one.
 func (l *logger) setMethodMethodLogger(method string, ml *MethodLoggerConfig) error {
 	if _, ok := l.config.Blacklist[method]; ok {
 		return fmt.Errorf("conflicting blacklist rules for method %v found", method)
@@ -161,11 +161,11 @@ func (l *logger) setBlacklist(method string) error {
 	return nil
 }
 
-// getMethodLogger returns the methodLogger for the given methodName.
+// getMethodLogger returns the MethodLogger for the given methodName.
 //
 // methodName should be in the format of "/service/method".
 //
-// Each methodLogger returned by this method is a new instance. This is to
+// Each MethodLogger returned by this method is a new instance. This is to
 // generate sequence id within the call.
 func (l *logger) GetMethodLogger(methodName string) MethodLogger {
 	s, m, err := grpcutil.ParseMethod(methodName)
@@ -174,16 +174,16 @@ func (l *logger) GetMethodLogger(methodName string) MethodLogger {
 		return nil
 	}
 	if ml, ok := l.config.Methods[s+"/"+m]; ok {
-		return newMethodLogger(ml.Header, ml.Message)
+		return NewTruncatingMethodLogger(ml.Header, ml.Message)
 	}
 	if _, ok := l.config.Blacklist[s+"/"+m]; ok {
 		return nil
 	}
 	if ml, ok := l.config.Services[s]; ok {
-		return newMethodLogger(ml.Header, ml.Message)
+		return NewTruncatingMethodLogger(ml.Header, ml.Message)
 	}
 	if l.config.All == nil {
 		return nil
 	}
-	return newMethodLogger(l.config.All.Header, l.config.All.Message)
+	return NewTruncatingMethodLogger(l.config.All.Header, l.config.All.Message)
 }

vendor/google.golang.org/grpc/internal/binarylog/env_config.go

@@ -30,15 +30,15 @@ import (
 // to build a new logger and assign it to binarylog.Logger.
 //
 // Example filter config strings:
-//  - "" Nothing will be logged
-//  - "*" All headers and messages will be fully logged.
-//  - "*{h}" Only headers will be logged.
-//  - "*{m:256}" Only the first 256 bytes of each message will be logged.
-//  - "Foo/*" Logs every method in service Foo
-//  - "Foo/*,-Foo/Bar" Logs every method in service Foo except method /Foo/Bar
-//  - "Foo/*,Foo/Bar{m:256}" Logs the first 256 bytes of each message in method
-//    /Foo/Bar, logs all headers and messages in every other method in service
-//    Foo.
+//   - "" Nothing will be logged
+//   - "*" All headers and messages will be fully logged.
+//   - "*{h}" Only headers will be logged.
+//   - "*{m:256}" Only the first 256 bytes of each message will be logged.
+//   - "Foo/*" Logs every method in service Foo
+//   - "Foo/*,-Foo/Bar" Logs every method in service Foo except method /Foo/Bar
+//   - "Foo/*,Foo/Bar{m:256}" Logs the first 256 bytes of each message in method
+//     /Foo/Bar, logs all headers and messages in every other method in service
+//     Foo.
 //
 // If two configs exist for one certain method or service, the one specified
 // later overrides the previous config.
@@ -57,7 +57,7 @@ func NewLoggerFromConfigString(s string) Logger {
 	return l
 }
 
-// fillMethodLoggerWithConfigString parses config, creates methodLogger and adds
+// fillMethodLoggerWithConfigString parses config, creates TruncatingMethodLogger and adds
 // it to the right map in the logger.
 func (l *logger) fillMethodLoggerWithConfigString(config string) error {
 	// "" is invalid.

vendor/google.golang.org/grpc/internal/binarylog/method_logger.go

@@ -19,6 +19,7 @@
 package binarylog
 
 import (
+	"context"
 	"net"
 	"strings"
 	"sync/atomic"
@@ -26,7 +27,7 @@ import (
 
 	"github.com/golang/protobuf/proto"
 	"github.com/golang/protobuf/ptypes"
-	pb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
+	binlogpb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
 )
@@ -49,10 +50,12 @@ var idGen callIDGenerator
 
 // MethodLogger is the sub-logger for each method.
 type MethodLogger interface {
-	Log(LogEntryConfig)
+	Log(context.Context, LogEntryConfig)
 }
 
-type methodLogger struct {
+// TruncatingMethodLogger is a method logger that truncates headers and messages
+// based on configured fields.
+type TruncatingMethodLogger struct {
 	headerMaxLen, messageMaxLen uint64
 
 	callID          uint64
@@ -61,8 +64,9 @@ type methodLogger struct {
 	sink Sink // TODO(blog): make this plugable.
 }
 
-func newMethodLogger(h, m uint64) *methodLogger {
-	return &methodLogger{
+// NewTruncatingMethodLogger returns a new truncating method logger.
+func NewTruncatingMethodLogger(h, m uint64) *TruncatingMethodLogger {
+	return &TruncatingMethodLogger{
 		headerMaxLen:  h,
 		messageMaxLen: m,
 
@@ -75,8 +79,8 @@ func newMethodLogger(h, m uint64) *methodLogger {
 
 // Build is an internal only method for building the proto message out of the
 // input event. It's made public to enable other library to reuse as much logic
-// in methodLogger as possible.
-func (ml *methodLogger) Build(c LogEntryConfig) *pb.GrpcLogEntry {
+// in TruncatingMethodLogger as possible.
+func (ml *TruncatingMethodLogger) Build(c LogEntryConfig) *binlogpb.GrpcLogEntry {
 	m := c.toProto()
 	timestamp, _ := ptypes.TimestampProto(time.Now())
 	m.Timestamp = timestamp
@@ -84,22 +88,22 @@ func (ml *methodLogger) Build(c LogEntryConfig) *pb.GrpcLogEntry {
 	m.SequenceIdWithinCall = ml.idWithinCallGen.next()
 
 	switch pay := m.Payload.(type) {
-	case *pb.GrpcLogEntry_ClientHeader:
+	case *binlogpb.GrpcLogEntry_ClientHeader:
 		m.PayloadTruncated = ml.truncateMetadata(pay.ClientHeader.GetMetadata())
-	case *pb.GrpcLogEntry_ServerHeader:
+	case *binlogpb.GrpcLogEntry_ServerHeader:
 		m.PayloadTruncated = ml.truncateMetadata(pay.ServerHeader.GetMetadata())
-	case *pb.GrpcLogEntry_Message:
+	case *binlogpb.GrpcLogEntry_Message:
 		m.PayloadTruncated = ml.truncateMessage(pay.Message)
 	}
 	return m
 }
 
 // Log creates a proto binary log entry, and logs it to the sink.
-func (ml *methodLogger) Log(c LogEntryConfig) {
+func (ml *TruncatingMethodLogger) Log(ctx context.Context, c LogEntryConfig) {
 	ml.sink.Write(ml.Build(c))
 }
 
-func (ml *methodLogger) truncateMetadata(mdPb *pb.Metadata) (truncated bool) {
+func (ml *TruncatingMethodLogger) truncateMetadata(mdPb *binlogpb.Metadata) (truncated bool) {
 	if ml.headerMaxLen == maxUInt {
 		return false
 	}
@@ -118,7 +122,7 @@ func (ml *methodLogger) truncateMetadata(mdPb *pb.Metadata) (truncated bool) {
 			// but not counted towards the size limit.
 			continue
 		}
-		currentEntryLen := uint64(len(entry.Value))
+		currentEntryLen := uint64(len(entry.GetKey())) + uint64(len(entry.GetValue()))
 		if currentEntryLen > bytesLimit {
 			break
 		}
@@ -129,7 +133,7 @@ func (ml *methodLogger) truncateMetadata(mdPb *pb.Metadata) (truncated bool) {
 	return truncated
 }
 
-func (ml *methodLogger) truncateMessage(msgPb *pb.Message) (truncated bool) {
+func (ml *TruncatingMethodLogger) truncateMessage(msgPb *binlogpb.Message) (truncated bool) {
 	if ml.messageMaxLen == maxUInt {
 		return false
 	}
@@ -142,7 +146,7 @@ func (ml *methodLogger) truncateMessage(msgPb *pb.Message) (truncated bool) {
 
 // LogEntryConfig represents the configuration for binary log entry.
 type LogEntryConfig interface {
-	toProto() *pb.GrpcLogEntry
+	toProto() *binlogpb.GrpcLogEntry
 }
 
 // ClientHeader configs the binary log entry to be a ClientHeader entry.
@@ -156,10 +160,10 @@ type ClientHeader struct {
 	PeerAddr net.Addr
 }
 
-func (c *ClientHeader) toProto() *pb.GrpcLogEntry {
+func (c *ClientHeader) toProto() *binlogpb.GrpcLogEntry {
 	// This function doesn't need to set all the fields (e.g. seq ID). The Log
 	// function will set the fields when necessary.
-	clientHeader := &pb.ClientHeader{
+	clientHeader := &binlogpb.ClientHeader{
 		Metadata:   mdToMetadataProto(c.Header),
 		MethodName: c.MethodName,
 		Authority:  c.Authority,
@@ -167,16 +171,16 @@ func (c *ClientHeader) toProto() *pb.GrpcLogEntry {
 	if c.Timeout > 0 {
 		clientHeader.Timeout = ptypes.DurationProto(c.Timeout)
 	}
-	ret := &pb.GrpcLogEntry{
-		Type: pb.GrpcLogEntry_EVENT_TYPE_CLIENT_HEADER,
-		Payload: &pb.GrpcLogEntry_ClientHeader{
+	ret := &binlogpb.GrpcLogEntry{
+		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_HEADER,
+		Payload: &binlogpb.GrpcLogEntry_ClientHeader{
 			ClientHeader: clientHeader,
 		},
 	}
 	if c.OnClientSide {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
 	} else {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
 	}
 	if c.PeerAddr != nil {
 		ret.Peer = addrToProto(c.PeerAddr)
@@ -192,19 +196,19 @@ type ServerHeader struct {
 	PeerAddr net.Addr
 }
 
-func (c *ServerHeader) toProto() *pb.GrpcLogEntry {
-	ret := &pb.GrpcLogEntry{
-		Type: pb.GrpcLogEntry_EVENT_TYPE_SERVER_HEADER,
-		Payload: &pb.GrpcLogEntry_ServerHeader{
-			ServerHeader: &pb.ServerHeader{
+func (c *ServerHeader) toProto() *binlogpb.GrpcLogEntry {
+	ret := &binlogpb.GrpcLogEntry{
+		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_HEADER,
+		Payload: &binlogpb.GrpcLogEntry_ServerHeader{
+			ServerHeader: &binlogpb.ServerHeader{
 				Metadata: mdToMetadataProto(c.Header),
 			},
 		},
 	}
 	if c.OnClientSide {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
 	} else {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
 	}
 	if c.PeerAddr != nil {
 		ret.Peer = addrToProto(c.PeerAddr)
@@ -220,7 +224,7 @@ type ClientMessage struct {
 	Message interface{}
 }
 
-func (c *ClientMessage) toProto() *pb.GrpcLogEntry {
+func (c *ClientMessage) toProto() *binlogpb.GrpcLogEntry {
 	var (
 		data []byte
 		err  error
@@ -235,19 +239,19 @@ func (c *ClientMessage) toProto() *pb.GrpcLogEntry {
 	} else {
 		grpclogLogger.Infof("binarylogging: message to log is neither proto.message nor []byte")
 	}
-	ret := &pb.GrpcLogEntry{
-		Type: pb.GrpcLogEntry_EVENT_TYPE_CLIENT_MESSAGE,
-		Payload: &pb.GrpcLogEntry_Message{
-			Message: &pb.Message{
+	ret := &binlogpb.GrpcLogEntry{
+		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_MESSAGE,
+		Payload: &binlogpb.GrpcLogEntry_Message{
+			Message: &binlogpb.Message{
 				Length: uint32(len(data)),
 				Data:   data,
 			},
 		},
 	}
 	if c.OnClientSide {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
 	} else {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
 	}
 	return ret
 }
@@ -260,7 +264,7 @@ type ServerMessage struct {
 	Message interface{}
 }
 
-func (c *ServerMessage) toProto() *pb.GrpcLogEntry {
+func (c *ServerMessage) toProto() *binlogpb.GrpcLogEntry {
 	var (
 		data []byte
 		err  error
@@ -275,19 +279,19 @@ func (c *ServerMessage) toProto() *pb.GrpcLogEntry {
 	} else {
 		grpclogLogger.Infof("binarylogging: message to log is neither proto.message nor []byte")
 	}
-	ret := &pb.GrpcLogEntry{
-		Type: pb.GrpcLogEntry_EVENT_TYPE_SERVER_MESSAGE,
-		Payload: &pb.GrpcLogEntry_Message{
-			Message: &pb.Message{
+	ret := &binlogpb.GrpcLogEntry{
+		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_MESSAGE,
+		Payload: &binlogpb.GrpcLogEntry_Message{
+			Message: &binlogpb.Message{
 				Length: uint32(len(data)),
 				Data:   data,
 			},
 		},
 	}
 	if c.OnClientSide {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
 	} else {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
 	}
 	return ret
 }
@@ -297,15 +301,15 @@ type ClientHalfClose struct {
 	OnClientSide bool
 }
 
-func (c *ClientHalfClose) toProto() *pb.GrpcLogEntry {
-	ret := &pb.GrpcLogEntry{
-		Type:    pb.GrpcLogEntry_EVENT_TYPE_CLIENT_HALF_CLOSE,
+func (c *ClientHalfClose) toProto() *binlogpb.GrpcLogEntry {
+	ret := &binlogpb.GrpcLogEntry{
+		Type:    binlogpb.GrpcLogEntry_EVENT_TYPE_CLIENT_HALF_CLOSE,
 		Payload: nil, // No payload here.
 	}
 	if c.OnClientSide {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
 	} else {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
 	}
 	return ret
 }
@@ -321,7 +325,7 @@ type ServerTrailer struct {
 	PeerAddr net.Addr
 }
 
-func (c *ServerTrailer) toProto() *pb.GrpcLogEntry {
+func (c *ServerTrailer) toProto() *binlogpb.GrpcLogEntry {
 	st, ok := status.FromError(c.Err)
 	if !ok {
 		grpclogLogger.Info("binarylogging: error in trailer is not a status error")
@@ -337,10 +341,10 @@ func (c *ServerTrailer) toProto() *pb.GrpcLogEntry {
 			grpclogLogger.Infof("binarylogging: failed to marshal status proto: %v", err)
 		}
 	}
-	ret := &pb.GrpcLogEntry{
-		Type: pb.GrpcLogEntry_EVENT_TYPE_SERVER_TRAILER,
-		Payload: &pb.GrpcLogEntry_Trailer{
-			Trailer: &pb.Trailer{
+	ret := &binlogpb.GrpcLogEntry{
+		Type: binlogpb.GrpcLogEntry_EVENT_TYPE_SERVER_TRAILER,
+		Payload: &binlogpb.GrpcLogEntry_Trailer{
+			Trailer: &binlogpb.Trailer{
 				Metadata:      mdToMetadataProto(c.Trailer),
 				StatusCode:    uint32(st.Code()),
 				StatusMessage: st.Message(),
@@ -349,9 +353,9 @@ func (c *ServerTrailer) toProto() *pb.GrpcLogEntry {
 		},
 	}
 	if c.OnClientSide {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
 	} else {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
 	}
 	if c.PeerAddr != nil {
 		ret.Peer = addrToProto(c.PeerAddr)
@@ -364,15 +368,15 @@ type Cancel struct {
 	OnClientSide bool
 }
 
-func (c *Cancel) toProto() *pb.GrpcLogEntry {
-	ret := &pb.GrpcLogEntry{
-		Type:    pb.GrpcLogEntry_EVENT_TYPE_CANCEL,
+func (c *Cancel) toProto() *binlogpb.GrpcLogEntry {
+	ret := &binlogpb.GrpcLogEntry{
+		Type:    binlogpb.GrpcLogEntry_EVENT_TYPE_CANCEL,
 		Payload: nil,
 	}
 	if c.OnClientSide {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_CLIENT
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_CLIENT
 	} else {
-		ret.Logger = pb.GrpcLogEntry_LOGGER_SERVER
+		ret.Logger = binlogpb.GrpcLogEntry_LOGGER_SERVER
 	}
 	return ret
 }
@@ -389,15 +393,15 @@ func metadataKeyOmit(key string) bool {
 	return strings.HasPrefix(key, "grpc-")
 }
 
-func mdToMetadataProto(md metadata.MD) *pb.Metadata {
-	ret := &pb.Metadata{}
+func mdToMetadataProto(md metadata.MD) *binlogpb.Metadata {
+	ret := &binlogpb.Metadata{}
 	for k, vv := range md {
 		if metadataKeyOmit(k) {
 			continue
 		}
 		for _, v := range vv {
 			ret.Entry = append(ret.Entry,
-				&pb.MetadataEntry{
+				&binlogpb.MetadataEntry{
 					Key:   k,
 					Value: []byte(v),
 				},
@@ -407,26 +411,26 @@ func mdToMetadataProto(md metadata.MD) *pb.Metadata {
 	return ret
 }
 
-func addrToProto(addr net.Addr) *pb.Address {
-	ret := &pb.Address{}
+func addrToProto(addr net.Addr) *binlogpb.Address {
+	ret := &binlogpb.Address{}
 	switch a := addr.(type) {
 	case *net.TCPAddr:
 		if a.IP.To4() != nil {
-			ret.Type = pb.Address_TYPE_IPV4
+			ret.Type = binlogpb.Address_TYPE_IPV4
 		} else if a.IP.To16() != nil {
-			ret.Type = pb.Address_TYPE_IPV6
+			ret.Type = binlogpb.Address_TYPE_IPV6
 		} else {
-			ret.Type = pb.Address_TYPE_UNKNOWN
+			ret.Type = binlogpb.Address_TYPE_UNKNOWN
 			// Do not set address and port fields.
 			break
 		}
 		ret.Address = a.IP.String()
 		ret.IpPort = uint32(a.Port)
 	case *net.UnixAddr:
-		ret.Type = pb.Address_TYPE_UNIX
+		ret.Type = binlogpb.Address_TYPE_UNIX
 		ret.Address = a.String()
 	default:
-		ret.Type = pb.Address_TYPE_UNKNOWN
+		ret.Type = binlogpb.Address_TYPE_UNKNOWN
 	}
 	return ret
 }

vendor/google.golang.org/grpc/internal/binarylog/sink.go

@@ -26,7 +26,7 @@ import (
 	"time"
 
 	"github.com/golang/protobuf/proto"
-	pb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
+	binlogpb "google.golang.org/grpc/binarylog/grpc_binarylog_v1"
 )
 
 var (
@@ -42,15 +42,15 @@ type Sink interface {
 	// Write will be called to write the log entry into the sink.
 	//
 	// It should be thread-safe so it can be called in parallel.
-	Write(*pb.GrpcLogEntry) error
+	Write(*binlogpb.GrpcLogEntry) error
 	// Close will be called when the Sink is replaced by a new Sink.
 	Close() error
 }
 
 type noopSink struct{}
 
-func (ns *noopSink) Write(*pb.GrpcLogEntry) error { return nil }
-func (ns *noopSink) Close() error                 { return nil }
+func (ns *noopSink) Write(*binlogpb.GrpcLogEntry) error { return nil }
+func (ns *noopSink) Close() error                       { return nil }
 
 // newWriterSink creates a binary log sink with the given writer.
 //
@@ -66,7 +66,7 @@ type writerSink struct {
 	out io.Writer
 }
 
-func (ws *writerSink) Write(e *pb.GrpcLogEntry) error {
+func (ws *writerSink) Write(e *binlogpb.GrpcLogEntry) error {
 	b, err := proto.Marshal(e)
 	if err != nil {
 		grpclogLogger.Errorf("binary logging: failed to marshal proto message: %v", err)
@@ -96,7 +96,7 @@ type bufferedSink struct {
 	done        chan struct{}
 }
 
-func (fs *bufferedSink) Write(e *pb.GrpcLogEntry) error {
+func (fs *bufferedSink) Write(e *binlogpb.GrpcLogEntry) error {
 	fs.mu.Lock()
 	defer fs.mu.Unlock()
 	if !fs.flusherStarted {