func (l *Librarian) Introduce(ctx context.Context, rq *api.IntroduceRequest) (
	*api.IntroduceResponse, error) {
	lg := l.logger.With(rqMetadataFields(rq.Metadata)...)
	lg.Debug("received introduce request", introduceRequestFields(rq)...)
	endpoint := api.Introduce
	requesterID, err := l.checkRequest(ctx, rq, rq.Metadata)
	if err != nil {
		l.record(requesterID, endpoint, comm.Request, comm.Error)
		return nil, logReturnInvalidRqErr(lg, err)
	}
	if err := l.allower.Allow(requesterID, endpoint); err != nil {
		l.record(requesterID, endpoint, comm.Request, comm.Error)
		return nil, logReturnNotAllowedErr(lg, err)
	}
	requester := l.fromer.FromAPI(rq.Self)
	if requester.ID().Cmp(requesterID) != 0 {
		l.record(requesterID, endpoint, comm.Request, comm.Error)
		return nil, logReturnInvalidRqErr(lg, errBadPeerIDSig)
	}
	l.record(requesterID, endpoint, comm.Request, comm.Success)
	l.rt.Push(requester)
	seed := int64(binary.BigEndian.Uint64(rq.Metadata.RequestId[:8]))
	peers := l.rt.Sample(uint(rq.NumPeers), rand.New(rand.NewSource(seed)))
	rp := &api.IntroduceResponse{
		Metadata: l.NewResponseMetadata(rq.Metadata),
		Self:     l.apiSelf,
		Peers:    peer.ToAPIs(peers),
	}
	lg.Info("introduced", introduceResponseFields(rp)...)
	return rp, nil
}
Introduce receives and gives identifying information about the peer in the network.
package server
import (
	"encoding/binary"
	"errors"
	"fmt"
	"math/rand"
	"net/http"
	"time"
	"github.com/drausin/libri/libri/common/db"
	"github.com/drausin/libri/libri/common/ecid"
	"github.com/drausin/libri/libri/common/id"
	"github.com/drausin/libri/libri/common/storage"
	"github.com/drausin/libri/libri/common/subscribe"
	"github.com/drausin/libri/libri/librarian/api"
	"github.com/drausin/libri/libri/librarian/client"
	"github.com/drausin/libri/libri/librarian/server/comm"
	"github.com/drausin/libri/libri/librarian/server/introduce"
	"github.com/drausin/libri/libri/librarian/server/peer"
	"github.com/drausin/libri/libri/librarian/server/replicate"
	"github.com/drausin/libri/libri/librarian/server/routing"
	"github.com/drausin/libri/libri/librarian/server/search"
	"github.com/drausin/libri/libri/librarian/server/store"
	"github.com/drausin/libri/libri/librarian/server/verify"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/willf/bloom"
	"go.uber.org/zap"
	"golang.org/x/net/context"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/health"
	"google.golang.org/grpc/status"
)
const (
	newPublicationsSlack = 16
)
var (
	errBadPeerIDSig           = errors.New("stated client peer ID does not match signature")
	errStoreUnexpectedResult  = errors.New("unexpected store result")
	errSearchUnexpectedResult = errors.New("unexpected search result")
)
type Librarian struct {
	peerID ecid.ID
	orgID ecid.ID
	config *Config
	apiSelf *api.PeerAddress
	introducer introduce.Introducer
	searcher search.Searcher
	storer store.Storer
	replicator replicate.Replicator
	subscribeFrom subscribe.From
	subscribeTo subscribe.To
	RecentPubs subscribe.RecentPublications
	rqv RequestVerifier
	db db.KVDB
	serverSL storage.StorerLoader
	documentSL storage.DocumentSL
	kc storage.Checker
	kvc storage.KeyValueChecker
	fromer peer.Fromer
	signer client.Signer
	clients client.Pool
	rt routing.Table
	storageMetrics *storageMetrics
	rec comm.QueryRecorder
	allower comm.Allower
	logger *zap.Logger
	health *health.Server
	metrics *http.Server
	stop chan struct{}
	stopped chan struct{}
}
func NewLibrarian(config *Config, logger *zap.Logger) (*Librarian, error) {
	rdb, err := db.NewRocksDB(config.DbDir)
	if err != nil {
		logger.Error("unable to init RocksDB", zap.Error(err))
		return nil, err
	}
	serverSL := storage.NewServerSL(rdb)
	documentSL := storage.NewDocumentSLD(rdb)
	peerID, err := loadOrCreatePeerID(logger, serverSL)
	if err != nil {
		return nil, err
	}
	selfLogger := logger.With(zap.String(logSelfIDShort, id.ShortHex(peerID.Bytes())))
	knower := comm.NewAlwaysKnower()
	windows := []time.Duration{comm.Second, comm.Day, comm.Week}
	recorder, getters := comm.NewWindowQueryRecorderGetters(knower, windows)
	if config.ReportMetrics {
		recorder = comm.NewPromScalarRecorder(peerID.ID(), recorder)
	}
	prefer := comm.NewRpPreferer(getters[comm.Day])
	allower := comm.NewDefaultAllower(knower, getters)
	doctor := comm.NewResponseTimeDoctor(getters[comm.Day])
	rt := routing.NewEmpty(peerID.ID(), prefer, doctor, config.Routing)
	clients, err := client.NewDefaultLRUPool()
	if err != nil {
		return nil, err
	}
	peerSigner := client.NewECDSASigner(peerID.Key())
	orgSigner := client.NewEmptySigner()
	if config.OrgID != nil {
		orgSigner = client.NewECDSASigner(config.OrgID.Key())
	}
	searcher := search.NewDefaultSearcher(peerSigner, orgSigner, recorder, doctor, clients)
	storer := store.NewStorer(peerSigner, orgSigner, recorder, doctor, searcher,
		client.NewStorerCreator(clients))
	introducer := introduce.NewDefaultIntroducer(peerSigner, orgSigner, recorder, peerID.ID(),
		clients)
	verifier := verify.NewDefaultVerifier(peerSigner, orgSigner, recorder, doctor, clients)
	newPubs := make(chan *subscribe.KeyedPub, newPublicationsSlack)
	recentPubs, err := subscribe.NewRecentPublications(config.SubscribeTo.RecentCacheSize)
	if err != nil {
		return nil, err
	}
	clientBalancer := routing.NewClientBalancer(rt, clients)
	subscribeTo := subscribe.NewTo(config.SubscribeTo, selfLogger, peerID, config.OrgID,
		clientBalancer, peerSigner, orgSigner, recentPubs, newPubs)
	metricsSM := http.NewServeMux()
	metricsSM.Handle("/metrics", promhttp.Handler())
	metrics := &http.Server{Addr: fmt.Sprintf(":%d", config.LocalMetricsPort), Handler: metricsSM}
	rng := rand.New(rand.NewSource(peerID.Int().Int64()))
	replicator := replicate.NewReplicator(
		peerID,
		config.OrgID,
		rt,
		documentSL,
		verifier,
		storer,
		config.Replicate,
		verify.NewDefaultParameters(),
		config.Store,
		rng,
		selfLogger,
	)
	storageMetrics := newStorageMetrics(serverSL)
	return &Librarian{
		peerID:         peerID,
		config:         config,
		apiSelf:        peer.FromAddress(peerID.ID(), config.PublicName, config.PublicAddr),
		introducer:     introducer,
		searcher:       searcher,
		replicator:     replicator,
		storer:         storer,
		subscribeFrom:  subscribe.NewFrom(config.SubscribeFrom, logger, newPubs),
		subscribeTo:    subscribeTo,
		RecentPubs:     recentPubs,
		rqv:            NewRequestVerifier(),
		db:             rdb,
		serverSL:       serverSL,
		documentSL:     documentSL,
		kc:             storage.NewExactLengthChecker(storage.EntriesKeyLength),
		kvc:            storage.NewHashKeyValueChecker(),
		fromer:         peer.NewFromer(),
		signer:         peerSigner,
		clients:        clients,
		rt:             rt,
		storageMetrics: storageMetrics,
		rec:            recorder,
		allower:        allower,
		logger:         selfLogger,
		health:         health.NewServer(),
		metrics:        metrics,
		stop:           make(chan struct{}),
		stopped:        make(chan struct{}),
	}, nil
}