func (c *Client) Send(ctx context.Context, r *pdu.SendReq) (*pdu.SendRes, io.ReadCloser, error) {
	ctx, endSpan := trace.WithSpan(ctx, "rpc.client.Send")
	defer endSpan()
	res, stream, err := c.dataClient.ReqSend(ctx, r)
	if err != nil {
		return nil, nil, err
	}
	if stream == nil {
		return res, nil, nil
	}
	return res, stream, nil
}
callers must ensure that the returned io.ReadCloser is closed
TODO expose dataClient interface to the outside world
package rpc
import (
	"context"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"sync/atomic"
	"time"
	"google.golang.org/grpc"
	"github.com/zrepl/zrepl/daemon/logging/trace"
	"github.com/google/uuid"
	"github.com/zrepl/zrepl/replication/logic"
	"github.com/zrepl/zrepl/replication/logic/pdu"
	"github.com/zrepl/zrepl/rpc/dataconn"
	"github.com/zrepl/zrepl/rpc/grpcclientidentity/grpchelper"
	"github.com/zrepl/zrepl/rpc/versionhandshake"
	"github.com/zrepl/zrepl/transport"
	"github.com/zrepl/zrepl/util/envconst"
)
type Client struct {
	dataClient    *dataconn.Client
	controlClient pdu.ReplicationClient  
	controlConn   *grpc.ClientConn
	loggers       Loggers
	closed        chan struct{}
}
var _ logic.Endpoint = &Client{}
var _ logic.Sender = &Client{}
var _ logic.Receiver = &Client{}
type DialContextFunc = func(ctx context.Context, network string, addr string) (net.Conn, error)
func NewClient(cn transport.Connecter, loggers Loggers) *Client {
	cn = versionhandshake.Connecter(cn, envconst.Duration("ZREPL_RPC_CLIENT_VERSIONHANDSHAKE_TIMEOUT", 10*time.Second))
	muxedConnecter := mux(cn)
	c := &Client{
		loggers: loggers,
		closed:  make(chan struct{}),
	}
	grpcConn := grpchelper.ClientConn(muxedConnecter.control, loggers.Control)
	go func() {
		ctx, cancel := context.WithCancel(context.Background())
		go func() {
			<-c.closed
			cancel()
		}()
		defer cancel()
		for ctx.Err() == nil {
			state := grpcConn.GetState()
			loggers.General.WithField("grpc_state", state.String()).Debug("grpc state change")
			grpcConn.WaitForStateChange(ctx, state)
		}
	}()
	c.controlClient = pdu.NewReplicationClient(grpcConn)
	c.controlConn = grpcConn
	c.dataClient = dataconn.NewClient(muxedConnecter.data, loggers.Data)
	return c
}
func (c *Client) Close() {
	close(c.closed)
	if err := c.controlConn.Close(); err != nil {
		c.loggers.General.WithError(err).Error("cannot close control connection")
	}
}