func (dht *IpfsDHT) sendRequest(ctx context.Context, p peer.ID, pmes *pb.Message) (*pb.Message, error) {
	ctx, _ = tag.New(ctx, metrics.UpsertMessageType(pmes))
	ms, err := dht.messageSenderForPeer(ctx, p)
	if err != nil {
		return nil, err
	}
	rpmes, err := ms.SendRequest(ctx, pmes)
	if err != nil {
		return nil, err
	}
	return rpmes, nil
}
sendRequest sends out a request, but also makes sure to
measure the RTT for latency measurements.
package main
import (
	"bufio"
	"context"
	"fmt"
	"io"
	"sync"
	"time"
	"github.com/libp2p/go-libp2p-core/helpers"
	"github.com/libp2p/go-libp2p-core/network"
	"github.com/libp2p/go-libp2p-core/peer"
	"github.com/libp2p/go-libp2p-kad-dht/metrics"
	pb "github.com/libp2p/go-libp2p-kad-dht/pb"
	ggio "github.com/gogo/protobuf/io"
	"github.com/libp2p/go-msgio"
	"go.opencensus.io/tag"
)
var dhtReadMessageTimeout = 10 * time.Second
var dhtStreamIdleTimeout = 1 * time.Minute
var ErrReadTimeout = fmt.Errorf("timed out reading response")
type bufferedDelimitedWriter struct {
	*bufio.Writer
	ggio.WriteCloser
}
var writerPool = sync.Pool{
	New: func() interface{} {
		w := bufio.NewWriter(nil)
		return &bufferedDelimitedWriter{
			Writer:      w,
			WriteCloser: ggio.NewDelimitedWriter(w),
		}
	},
}
func writeMsg(w io.Writer, mes *pb.Message) error {
	bw := writerPool.Get().(*bufferedDelimitedWriter)
	bw.Reset(w)
	err := bw.WriteMsg(mes)
	if err == nil {
		err = bw.Flush()
	}
	bw.Reset(nil)
	writerPool.Put(bw)
	return err
}
func (w *bufferedDelimitedWriter) Flush() error {
	return w.Writer.Flush()
}