func (s *Store) applyProtoMessage(m proto.Message) error {
	cmd, err := proto.Marshal(m)
	if err != nil {
		return err
	}
	return s.raft.Apply(cmd, raftTimeout).Error()
}
applyProtoMessage applies a proto message.
package store
import (
	"context"
	"fmt"
	"os"
	"path"
	"path/filepath"
	"time"
	"github.com/casbin/hraft-dispatcher/store/logstore"
	"github.com/cenkalti/backoff/v4"
	"github.com/pkg/errors"
	"github.com/casbin/casbin/v2"
	"github.com/casbin/hraft-dispatcher/command"
	"github.com/casbin/hraft-dispatcher/http"
	"google.golang.org/protobuf/proto"
	"github.com/hashicorp/go-multierror"
	"github.com/hashicorp/raft"
	"go.uber.org/zap"
)
const (
	raftDBName          = "raft.db"
	retainSnapshotCount = 2
	raftTimeout         = 10 * time.Second
)
var _ http.Store = &Store{}
type Store struct {
	dataDir       string
	serverAddress string
	serverID      string
	raft                   *raft.Raft
	raftConfig             *raft.Config
	networkTransportConfig *raft.NetworkTransportConfig
	transport              raft.Transport
	snapshotStore          raft.SnapshotStore
	logStore               raft.LogStore
	stableStore            raft.StableStore
	fms                    raft.FSM
	boltStore              *logstore.BoltStore
	enforcer casbin.IDistributedEnforcer
	inMemory bool
	logger *zap.Logger
}
type Config struct {
	ID                     string
	Dir                    string
	NetworkTransportConfig *raft.NetworkTransportConfig
	Enforcer               casbin.IDistributedEnforcer
	RaftConfig             *raft.Config
}
func NewStore(logger *zap.Logger, config *Config) (*Store, error) {
	s := &Store{
		dataDir:                config.Dir,
		serverID:               config.ID,
		logger:                 logger,
		networkTransportConfig: config.NetworkTransportConfig,
		enforcer:               config.Enforcer,
		raftConfig:             config.RaftConfig,
	}
	return s, nil
}
func (s *Store) Start(enableBootstrap bool) error {
	var config *raft.Config
	if s.raftConfig == nil {
		config = raft.DefaultConfig()
		s.raftConfig = config
	} else {
		config = s.raftConfig
	}
	if len(config.LocalID) == 0 {
		config.LocalID = raft.ServerID(s.serverID)
	}
	var transport raft.Transport
	if s.inMemory {
		_, transport = raft.NewInmemTransport(raft.ServerAddress(s.serverAddress))
	} else {
		transport = raft.NewNetworkTransportWithConfig(s.networkTransportConfig)
	}
	s.transport = transport
	var snapshots raft.SnapshotStore
	if s.inMemory {
		snapshots = raft.NewInmemSnapshotStore()
	} else {
		fileSnapshots, err := raft.NewFileSnapshotStore(s.dataDir, retainSnapshotCount, os.Stderr)
		if err != nil {
			s.logger.Error("failed to new file snapshot store", zap.Error(err), zap.String("raftData", s.dataDir))
			return err
		}
		snapshots = fileSnapshots
	}
	s.snapshotStore = snapshots
	if s.inMemory {
		inMemStore := raft.NewInmemStore()
		s.logStore = inMemStore
		s.stableStore = inMemStore
	} else {
		dbPath := filepath.Join(s.dataDir, raftDBName)
		boltDB, err := logstore.NewBoltStore(dbPath)
		if err != nil {
			s.logger.Error("failed to new bolt store", zap.Error(err), zap.String("path", dbPath))
			return err
		}
		s.boltStore = boltDB
		s.logStore = boltDB
		s.stableStore = boltDB
	}
	fsm, err := NewFSM(s.logger, s.dataDir, s.enforcer)
	if err != nil {
		s.logger.Error("failed to new fsm", zap.Error(err))
		return err
	}
	ra, err := raft.NewRaft(config, fsm, s.logStore, s.stableStore, s.snapshotStore, s.transport)
	if err != nil {
		s.logger.Error("failed to new raft", zap.Error(err))
		return err
	}
	s.raft = ra
	if enableBootstrap {
		configuration := raft.Configuration{
			Servers: []raft.Server{
				{
					ID:      raft.ServerID(s.serverID),
					Address: s.transport.LocalAddr(),
				},
			},
		}
		f := ra.BootstrapCluster(configuration)
		if f.Error() != nil && f.Error() != raft.ErrCantBootstrap {
			s.logger.Error("failed to boostrap cluster", zap.Error(f.Error()))
			return f.Error()
		}
	}
	s.logger.Info(fmt.Sprintf("listening and serving Raft on %s", transport.LocalAddr()))
	return nil
}
func (s *Store) Stop() error {
	var result error
	shutdown := s.raft.Shutdown()
	if shutdown.Error() != nil {
		s.logger.Error("failed to stop the raft server", zap.Error(shutdown.Error()))
		result = multierror.Append(result, shutdown.Error())
	}
	if !s.inMemory {
		err := s.boltStore.Close()
		if err != nil {
			s.logger.Error("failed to close bolt database", zap.Error(err))
			result = multierror.Append(result, err)
		}
	}
	err := s.networkTransportConfig.Stream.Close()
	if err != nil {
		result = multierror.Append(result, err)
	}
	return result
}
func (s *Store) IsInitializedCluster() bool {
	if _, err := os.Stat(path.Join(s.dataDir, raftDBName)); err != nil {
		if os.IsNotExist(err) {
			return false
		}
	}
	return true
}
func (s *Store) WaitLeader() error {
	if s.raft.Leader() != "" {
		return nil
	}
	ticker := backoff.NewTicker(backoff.NewExponentialBackOff())
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()
	for {
		select {
		case <-ctx.Done():
			ticker.Stop()
			return errors.New("failed to detect the current leader")
		case <-ticker.C:
			if s.raft.Leader() != "" {
				ticker.Stop()
				return nil
			}
		}
	}
}
func (s *Store) Address() string {
	return s.networkTransportConfig.Stream.Addr().String()
}
func (s *Store) ID() string {
	return s.serverID
}
func (s *Store) DataDir() string {
	return s.dataDir
}