func NewBlockReplicator(conf *Config) (*BlockReplicator, error) {
	raftID, err := comm.MemberRaftID(conf.LocalConf.Server.Identity.ID, conf.ClusterConfig)
	if err != nil {
		return nil, err
	}
	lg := conf.Logger.With("nodeID", conf.LocalConf.Server.Identity.ID, "raftID", raftID)
	haveWAL := wal.Exist(conf.LocalConf.Replication.WALDir)
	storage, err := CreateStorage(lg, conf.LocalConf.Replication.WALDir, conf.LocalConf.Replication.SnapDir)
	if err != nil {
		return nil, errors.Errorf("failed to restore persisted raft data: %s", err)
	}
	storage.SnapshotCatchUpEntries = DefaultSnapshotCatchUpEntries
	var snapBlkNum uint64
	var confState raftpb.ConfState
	if s := storage.Snapshot(); !raft.IsEmptySnap(s) {
		snapBlock := &types.Block{}
		if err := proto.Unmarshal(s.Data, snapBlock); err != nil {
			return nil, errors.Wrapf(err, "failed to unmarshal snapshot block")
		}
		snapBlkNum = snapBlock.GetHeader().GetBaseHeader().GetNumber()
		confState = s.Metadata.ConfState
		lg.Debugf("Starting from last snapshot: block number [%d], Raft ConfState: %+v", snapBlkNum, confState)
	}
	br := &BlockReplicator{
		localConf:        conf.LocalConf,
		proposeCh:        make(chan *types.Block, 1),
		raftID:           raftID,
		raftStorage:      storage,
		oneQueueBarrier:  conf.BlockOneQueueBarrier,
		stopCh:           make(chan struct{}),
		doneProposeCh:    make(chan struct{}),
		doneEventCh:      make(chan struct{}),
		clusterConfig:    conf.ClusterConfig,
		transport:        conf.Transport,
		ledgerReader:     conf.LedgerReader,
		pendingTxs:       conf.PendingTxs,
		sizeLimit:        conf.ClusterConfig.ConsensusConfig.RaftConfig.SnapshotIntervalSize,
		confState:        confState,
		lastSnapBlockNum: snapBlkNum,
		lg:               lg,
	}
	height, err := br.ledgerReader.Height()
	if err != nil {
		br.lg.Panic("Failed to read block height")
	}
	if height > 0 {
		br.lastBlock, err = br.ledgerReader.Get(height)
		if err != nil {
			br.lg.Panic("Failed to read last block")
		}
	}
	if height > 1 {
		metadata := br.lastBlock.GetConsensusMetadata()
		br.appliedIndex = metadata.GetRaftIndex()
		br.lg.Debugf("last block [%d], consensus metadata: %+v", height, metadata)
	}
	raftConfig := &raft.Config{
		ID:              raftID,
		ElectionTick:    int(br.clusterConfig.ConsensusConfig.RaftConfig.ElectionTicks),
		HeartbeatTick:   int(br.clusterConfig.ConsensusConfig.RaftConfig.HeartbeatTicks),
		MaxSizePerMsg:   br.localConf.BlockCreation.MaxBlockSize,
		MaxInflightMsgs: int(br.clusterConfig.ConsensusConfig.RaftConfig.MaxInflightBlocks),
		Logger:          lg,
		Storage:         br.raftStorage.MemoryStorage,
		PreVote:                   true,
		CheckQuorum:               true,
		DisableProposalForwarding: true,  
	}
	lg.Debugf("haveWAL: %v, Storage: %v, Raft config: %+v", haveWAL, storage, raftConfig)
	joinExistingCluster := false  
	if haveWAL {
		br.raftNode = raft.RestartNode(raftConfig)
	} else {
		if joinExistingCluster {
			br.lg.Panicf("not supported yet: joinExistingCluster")
		} else {
			startPeers := raftPeers(br.clusterConfig)
			br.raftNode = raft.StartNode(raftConfig, startPeers)
		}
	}
	return br, nil
}
NewBlockReplicator creates a new BlockReplicator.
package replication
import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/hyperledger-labs/orion-server/config"
	"github.com/hyperledger-labs/orion-server/internal/comm"
	ierrors "github.com/hyperledger-labs/orion-server/internal/errors"
	"github.com/hyperledger-labs/orion-server/internal/queue"
	"github.com/hyperledger-labs/orion-server/pkg/logger"
	"github.com/hyperledger-labs/orion-server/pkg/types"
	"github.com/golang/protobuf/proto"
	"github.com/pkg/errors"
	"go.etcd.io/etcd/raft"
	"go.etcd.io/etcd/raft/raftpb"
	"go.etcd.io/etcd/wal"
)
const (
	DefaultSnapshotCatchUpEntries = uint64(4)
)
type BlockLedgerReader interface {
	Height() (uint64, error)
	Get(blockNumber uint64) (*types.Block, error)
}
type BlockReplicator struct {
	localConf *config.LocalConfiguration
	proposeCh       chan *types.Block
	raftID          uint64
	raftStorage     *RaftStorage
	raftNode        raft.Node
	oneQueueBarrier *queue.OneQueueBarrier  
	transport       *comm.HTTPTransport
	ledgerReader    BlockLedgerReader
	pendingTxs      *queue.PendingTxs  
	stopCh        chan struct{}
	stopOnce      sync.Once
	doneProposeCh chan struct{}
	doneEventCh   chan struct{}
	mutex               sync.Mutex
	clusterConfig       *types.ClusterConfig
	lastKnownLeader     uint64
	lastKnownLeaderHost string  
	appliedIndex uint64
	lastBlock    *types.Block
	sizeLimit        uint64  
	accDataSize      uint64  
	lastSnapBlockNum uint64
	confState        raftpb.ConfState  
	lg *logger.SugarLogger
}
type Config struct {
	LocalConf            *config.LocalConfiguration
	ClusterConfig        *types.ClusterConfig
	LedgerReader         BlockLedgerReader
	Transport            *comm.HTTPTransport
	BlockOneQueueBarrier *queue.OneQueueBarrier
	PendingTxs           *queue.PendingTxs
	Logger               *logger.SugarLogger
}