func startPeerConnections(ownID, n uint32, connector api.ReplicaConnector, handleMessage messageHandler, logger logger.Logger) error {
	for peerID := uint32(0); peerID < n; peerID++ {
		if peerID == ownID {
			continue
		}
		remote := fmt.Sprintf("replica %d", peerID)
		connect := makePeerConnector(peerID, connector)
		handleReplyStream := makeMessageStreamHandler(handleMessage, remote, logger)
		if err := startPeerConnection(ownID, connect, handleReplyStream); err != nil {
			return fmt.Errorf("cannot connect to replica %d: %s", peerID, err)
		}
	}
	return nil
}
startPeerConnections initiates asynchronous message exchange with
peer replicas.
package minbft
import (
	"fmt"
	"sync"
	"github.com/hyperledger-labs/minbft/api"
	"github.com/hyperledger-labs/minbft/common/logger"
	"github.com/hyperledger-labs/minbft/core/internal/clientstate"
	"github.com/hyperledger-labs/minbft/core/internal/messagelog"
	"github.com/hyperledger-labs/minbft/core/internal/peerstate"
	"github.com/hyperledger-labs/minbft/core/internal/requestlist"
	"github.com/hyperledger-labs/minbft/core/internal/viewstate"
	"github.com/hyperledger-labs/minbft/messages"
)
type messageStreamHandler func(in <-chan []byte, reply chan<- []byte)
type messageHandler func(msg messages.Message) (reply <-chan messages.Message, new bool, err error)
type peerConnector func(out <-chan []byte) (in <-chan []byte, err error)
type messageValidator func(msg messages.Message) error
type messageProcessor func(msg messages.Message) (new bool, err error)
type peerMessageProcessor func(msg messages.PeerMessage) (new bool, err error)
type embeddedMessageProcessor func(msg messages.PeerMessage)
type uiMessageProcessor func(msg messages.CertifiedMessage) (new bool, err error)
type viewMessageProcessor func(msg messages.PeerMessage) (new bool, err error)
type peerMessageApplier func(msg messages.PeerMessage, active bool) error
type generatedMessageHandler func(msg messages.ReplicaMessage)
type generatedMessageConsumer func(msg messages.ReplicaMessage)
func defaultMessageHandlers(id uint32, log messagelog.MessageLog, unicastLogs map[uint32]messagelog.MessageLog, config api.Configer, stack Stack, logger logger.Logger) (handleOwnMessage, handlePeerMessage, handleClientMessage messageHandler) {
	n := config.N()
	f := config.F()
	reqTimeout := makeRequestTimeoutProvider(config)
	prepTimeout := makePrepareTimeoutProvider(config)
	verifyMessageSignature := makeMessageSignatureVerifier(stack, messages.AuthenBytes)
	signMessage := makeMessageSigner(stack, messages.AuthenBytes)
	verifyUI := makeUIVerifier(stack, messages.AuthenBytes)
	assignUI := makeUIAssigner(stack, messages.AuthenBytes)
	clientStates := clientstate.NewProvider(reqTimeout, prepTimeout)
	peerStates := peerstate.NewProvider()
	viewState := viewstate.New()
	captureSeq := makeRequestSeqCapturer(clientStates)
	prepareSeq := makeRequestSeqPreparer(clientStates)
	retireSeq := makeRequestSeqRetirer(clientStates)
	pendingReq := requestlist.New()
	captureUI := makeUICapturer(peerStates)
	consumeGeneratedMessage := makeGeneratedMessageConsumer(log, clientStates, logger)
	handleGeneratedMessage := makeGeneratedMessageHandler(signMessage, assignUI, consumeGeneratedMessage)
	requestViewChange := makeViewChangeRequestor(id, viewState, handleGeneratedMessage)
	handleReqTimeout := makeRequestTimeoutHandler(requestViewChange, logger)
	startReqTimer := makeRequestTimerStarter(clientStates, handleReqTimeout, logger)
	stopReqTimer := makeRequestTimerStopper(clientStates)
	startPrepTimer := makePrepareTimerStarter(n, clientStates, unicastLogs, logger)
	stopPrepTimer := makePrepareTimerStopper(clientStates)
	acceptCommitment := makeCommitmentAcceptor()
	countCommitment := makeCommitmentCounter(f)
	executeRequest := makeRequestExecutor(id, retireSeq, pendingReq, stopReqTimer, stack, handleGeneratedMessage)
	collectCommitment := makeCommitmentCollector(acceptCommitment, countCommitment, executeRequest)
	validateRequest := makeRequestValidator(verifyMessageSignature)
	validatePrepare := makePrepareValidator(n, verifyUI, validateRequest)
	validateCommit := makeCommitValidator(verifyUI, validatePrepare)
	validateReqViewChange := makeReqViewChangeValidator(verifyMessageSignature)
	validateMessage := makeMessageValidator(validateRequest, validatePrepare, validateCommit, validateReqViewChange)
	applyCommit := makeCommitApplier(collectCommitment)
	applyPrepare := makePrepareApplier(id, prepareSeq, collectCommitment, handleGeneratedMessage, stopPrepTimer)
	applyPeerMessage := makePeerMessageApplier(applyPrepare, applyCommit)
	applyRequest := makeRequestApplier(id, n, handleGeneratedMessage, startReqTimer, startPrepTimer)
	var processMessage messageProcessor
	processMessageThunk := func(msg messages.Message) (new bool, err error) {
		return processMessage(msg)
	}
	processRequest := makeRequestProcessor(captureSeq, pendingReq, viewState, applyRequest)
	processViewMessage := makeViewMessageProcessor(viewState, applyPeerMessage)
	processUIMessage := makeUIMessageProcessor(captureUI, processViewMessage)
	processEmbedded := makeEmbeddedMessageProcessor(processMessageThunk, logger)
	collectReqViewChange := makeReqViewChangeCollector(f)
	startViewChange := makeViewChangeStarter(id, viewState, log, handleGeneratedMessage)
	processReqViewChange := makeReqViewChangeProcessor(collectReqViewChange, startViewChange)
	processPeerMessage := makePeerMessageProcessor(processEmbedded, processUIMessage, processReqViewChange)
	processMessage = makeMessageProcessor(processRequest, processPeerMessage)
	handleOwnMessage = makeOwnMessageHandler(processMessage)
	handlePeerMessage = makePeerMessageHandler(validateMessage, processMessage)
	replyRequest := makeRequestReplier(clientStates)
	handleClientMessage = makeClientMessageHandler(validateRequest, processRequest, replyRequest)
	return
}
func makeMessageStreamHandler(handleMessage messageHandler, remote string, logger logger.Logger) messageStreamHandler {
	return func(in <-chan []byte, out chan<- []byte) {
		for msgBytes := range in {
			msg, err := messageImpl.NewFromBinary(msgBytes)
			if err != nil {
				logger.Warningf("Error unmarshaling message from %s: %s", remote, err)
				return
			}
			msgStr := messages.Stringify(msg)
			logger.Debugf("Received %s from %s", msgStr, remote)
			replyChan, new, err := handleMessage(msg)
			if err != nil {
				logger.Warningf("Error handling %s from %s: %s", msgStr, remote, err)
				return
			} else if !new {
				logger.Debugf("Dropped %s from %s", msgStr, remote)
			} else {
				logger.Debugf("Handled %s from %s", msgStr, remote)
			}
			if replyChan != nil {
				remote := remote  
				switch m := msg.(type) {
				case messages.Hello:
					remote = fmt.Sprintf("replica %d", m.ReplicaID())
				case messages.ClientMessage:
					remote = fmt.Sprintf("client %d", m.ClientID())
				}
				for m := range replyChan {
					mStr := messages.Stringify(m)
					logger.Debugf("Sending %s to %s", mStr, remote)
					replyBytes, err := m.MarshalBinary()
					if err != nil {
						panic(err)
					}
					out <- replyBytes
				}
			}
		}
	}
}