func (d *Handler) beforeSyncStateGetSnapshot(e *fsm.Event) {
	peerLogger.Debug("Received message: %s", e.Event)
	msg, ok := e.Args[0].(*pb.OpenchainMessage)
	if !ok {
		e.Cancel(fmt.Errorf("Received unexpected message type"))
		return
	}
	syncStateSnapshotRequest := &pb.SyncStateSnapshotRequest{}
	err := proto.Unmarshal(msg.Payload, syncStateSnapshotRequest)
	if err != nil {
		e.Cancel(fmt.Errorf("Error unmarshalling SyncStateSnapshotRequest in beforeSyncStateGetSnapshot: %s", err))
		return
	}
	go d.sendStateSnapshot(syncStateSnapshotRequest)
}
beforeSyncStateGetSnapshot triggers the sending of State Snapshot deltas to remote Peer.
package peer
import (
	"fmt"
	"sync"
	"time"
	"github.com/golang/protobuf/proto"
	"github.com/looplab/fsm"
	"github.com/spf13/viper"
	"github.com/openblockchain/obc-peer/openchain/ledger/statemgmt"
	pb "github.com/openblockchain/obc-peer/protos"
)
type Handler struct {
	chatMutex	               sync.Mutex
	ToPeerEndpoint                *pb.PeerEndpoint
	Coordinator                   MessageHandlerCoordinator
	ChatStream                    ChatStream
	doneChan                      chan struct{}
	FSM                           *fsm.FSM
	initiatedStream               bool  
	registered                    bool
	syncBlocksMutex               sync.Mutex
	syncBlocks                    chan *pb.SyncBlocks
	snapshotRequestHandler        *syncStateSnapshotRequestHandler
	syncStateDeltasRequestHandler *syncStateDeltasHandler
}
func NewPeerHandler(coord MessageHandlerCoordinator, stream ChatStream, initiatedStream bool, nextHandler MessageHandler) (MessageHandler, error) {
	d := &Handler{
		ChatStream:      stream,
		initiatedStream: initiatedStream,
		Coordinator:     coord,
	}
	d.doneChan = make(chan struct{})
	d.snapshotRequestHandler = newSyncStateSnapshotRequestHandler()
	d.syncStateDeltasRequestHandler = newSyncStateDeltasHandler()
	d.FSM = fsm.NewFSM(
		"created",
		fsm.Events{
			{Name: pb.OpenchainMessage_DISC_HELLO.String(), Src: []string{"created"}, Dst: "established"},
			{Name: pb.OpenchainMessage_DISC_GET_PEERS.String(), Src: []string{"established"}, Dst: "established"},
			{Name: pb.OpenchainMessage_DISC_PEERS.String(), Src: []string{"established"}, Dst: "established"},
			{Name: pb.OpenchainMessage_SYNC_BLOCK_ADDED.String(), Src: []string{"established"}, Dst: "established"},
			{Name: pb.OpenchainMessage_SYNC_GET_BLOCKS.String(), Src: []string{"established"}, Dst: "established"},
			{Name: pb.OpenchainMessage_SYNC_BLOCKS.String(), Src: []string{"established"}, Dst: "established"},
			{Name: pb.OpenchainMessage_SYNC_STATE_GET_SNAPSHOT.String(), Src: []string{"established"}, Dst: "established"},
			{Name: pb.OpenchainMessage_SYNC_STATE_SNAPSHOT.String(), Src: []string{"established"}, Dst: "established"},
			{Name: pb.OpenchainMessage_SYNC_STATE_GET_DELTAS.String(), Src: []string{"established"}, Dst: "established"},
			{Name: pb.OpenchainMessage_SYNC_STATE_DELTAS.String(), Src: []string{"established"}, Dst: "established"},
		},
		fsm.Callbacks{
			"enter_state":                                                    func(e *fsm.Event) { d.enterState(e) },
			"before_" + pb.OpenchainMessage_DISC_HELLO.String():              func(e *fsm.Event) { d.beforeHello(e) },
			"before_" + pb.OpenchainMessage_DISC_GET_PEERS.String():          func(e *fsm.Event) { d.beforeGetPeers(e) },
			"before_" + pb.OpenchainMessage_DISC_PEERS.String():              func(e *fsm.Event) { d.beforePeers(e) },
			"before_" + pb.OpenchainMessage_SYNC_BLOCK_ADDED.String():        func(e *fsm.Event) { d.beforeBlockAdded(e) },
			"before_" + pb.OpenchainMessage_SYNC_GET_BLOCKS.String():         func(e *fsm.Event) { d.beforeSyncGetBlocks(e) },
			"before_" + pb.OpenchainMessage_SYNC_BLOCKS.String():             func(e *fsm.Event) { d.beforeSyncBlocks(e) },
			"before_" + pb.OpenchainMessage_SYNC_STATE_GET_SNAPSHOT.String(): func(e *fsm.Event) { d.beforeSyncStateGetSnapshot(e) },
			"before_" + pb.OpenchainMessage_SYNC_STATE_SNAPSHOT.String():     func(e *fsm.Event) { d.beforeSyncStateSnapshot(e) },
			"before_" + pb.OpenchainMessage_SYNC_STATE_GET_DELTAS.String():   func(e *fsm.Event) { d.beforeSyncStateGetDeltas(e) },
			"before_" + pb.OpenchainMessage_SYNC_STATE_DELTAS.String():       func(e *fsm.Event) { d.beforeSyncStateDeltas(e) },
		},
	)
	if d.initiatedStream {
		helloMessage, err := d.Coordinator.NewOpenchainDiscoveryHello()
		if err != nil {
			return nil, fmt.Errorf("Error getting new HelloMessage: %s", err)
		}
		if err := d.SendMessage(helloMessage); err != nil {
			return nil, fmt.Errorf("Error creating new Peer Handler, error returned sending %s: %s", pb.OpenchainMessage_DISC_HELLO, err)
		}
	}
	return d, nil
}
func (d *Handler) enterState(e *fsm.Event) {
	peerLogger.Debug("The Peer's bi-directional stream to %s is %s, from event %s\n", d.ToPeerEndpoint, e.Dst, e.Event)
}
func (d *Handler) deregister() error {
	var err error
	if d.registered {
		err = d.Coordinator.DeregisterHandler(d)
		d.doneChan <- struct{}{}
		d.registered = false
	}
	return err
}
func (d *Handler) To() (pb.PeerEndpoint, error) {
	if d.ToPeerEndpoint == nil {
		return pb.PeerEndpoint{}, fmt.Errorf("No peer endpoint for handler")
	}
	return *(d.ToPeerEndpoint), nil
}
func (d *Handler) Stop() error {
	err := d.deregister()
	if err != nil {
		return fmt.Errorf("Error stopping MessageHandler: %s", err)
	}
	return nil
}
func (d *Handler) beforeHello(e *fsm.Event) {
	peerLogger.Debug("Received %s, parsing out Peer identification", e.Event)
	if _, ok := e.Args[0].(*pb.OpenchainMessage); !ok {
		e.Cancel(fmt.Errorf("Received unexpected message type"))
		return
	}
	msg := e.Args[0].(*pb.OpenchainMessage)
	helloMessage := &pb.HelloMessage{}
	err := proto.Unmarshal(msg.Payload, helloMessage)
	if err != nil {
		e.Cancel(fmt.Errorf("Error unmarshalling HelloMessage: %s", err))
		return
	}
	d.ToPeerEndpoint = helloMessage.PeerEndpoint
	peerLogger.Debug("Received %s from endpoint=%s", e.Event, helloMessage)
	if viper.GetBool("security.enabled") {
		if err := d.Coordinator.GetSecHelper().Verify(helloMessage.PeerEndpoint.PkiID, msg.Signature, msg.Payload); err != nil {
			e.Cancel(fmt.Errorf("Error Verifying signature for received HelloMessage: %s", err))
			return
		}
		peerLogger.Debug("Verified signature for %s", e.Event)
	}
	if d.initiatedStream == false {
		peerLogger.Debug("Received %s, sending back %s", e.Event, pb.OpenchainMessage_DISC_HELLO.String())
		helloMessage, err := d.Coordinator.NewOpenchainDiscoveryHello()
		if err != nil {
			e.Cancel(fmt.Errorf("Error getting new HelloMessage: %s", err))
			return
		}
		if err := d.SendMessage(helloMessage); err != nil {
			e.Cancel(fmt.Errorf("Error sending response to %s:  %s", e.Event, err))
			return
		}
	}
	err = d.Coordinator.RegisterHandler(d)
	if err != nil {
		e.Cancel(fmt.Errorf("Error registering Handler: %s", err))
	} else {
		d.registered = true
		go d.start()
	}
}
func (d *Handler) beforeGetPeers(e *fsm.Event) {
	peersMessage, err := d.Coordinator.GetPeers()
	if err != nil {
		e.Cancel(fmt.Errorf("Error Getting Peers: %s", err))
		return
	}
	data, err := proto.Marshal(peersMessage)
	if err != nil {
		e.Cancel(fmt.Errorf("Error Marshalling PeersMessage: %s", err))
		return
	}
	peerLogger.Debug("Sending back %s", pb.OpenchainMessage_DISC_PEERS.String())
	if err := d.SendMessage(&pb.OpenchainMessage{Type: pb.OpenchainMessage_DISC_PEERS, Payload: data}); err != nil {
		e.Cancel(err)
	}
}
func (d *Handler) beforePeers(e *fsm.Event) {
	peerLogger.Debug("Received %s, grabbing peers message", e.Event)
	if _, ok := e.Args[0].(*pb.OpenchainMessage); !ok {
		e.Cancel(fmt.Errorf("Received unexpected message type"))
		return
	}
	msg := e.Args[0].(*pb.OpenchainMessage)
	peersMessage := &pb.PeersMessage{}
	err := proto.Unmarshal(msg.Payload, peersMessage)
	if err != nil {
		e.Cancel(fmt.Errorf("Error unmarshalling PeersMessage: %s", err))
		return
	}
	peerLogger.Debug("Received PeersMessage with Peers: %s", peersMessage)
	d.Coordinator.PeersDiscovered(peersMessage)
}
func (d *Handler) beforeBlockAdded(e *fsm.Event) {
	peerLogger.Debug("Received message: %s", e.Event)
	msg, ok := e.Args[0].(*pb.OpenchainMessage)
	if !ok {
		e.Cancel(fmt.Errorf("Received unexpected message type"))
		return
	}
	_ = msg
}
func (d *Handler) when(stateToCheck string) bool {
	return d.FSM.Is(stateToCheck)
}
func (d *Handler) HandleMessage(msg *pb.OpenchainMessage) error {
	peerLogger.Debug("Handling OpenchainMessage of type: %s ", msg.Type)
	if d.FSM.Cannot(msg.Type.String()) {
		return fmt.Errorf("Peer FSM cannot handle message (%s) with payload size (%d) while in state: %s", msg.Type.String(), len(msg.Payload), d.FSM.Current())
	}
	err := d.FSM.Event(msg.Type.String(), msg)
	if err != nil {
		if _, ok := err.(*fsm.NoTransitionError); !ok {
			return fmt.Errorf("Peer FSM failed while handling message (%s): current state: %s, error: %s", msg.Type.String(), d.FSM.Current(), err)
		}
	}
	return nil
}
func (d *Handler) SendMessage(msg *pb.OpenchainMessage) error {
	d.chatMutex.Lock()
	defer d.chatMutex.Unlock()
	peerLogger.Debug("Sending message to stream of type: %s ", msg.Type)
	err := d.ChatStream.Send(msg)
	if err != nil {
		return fmt.Errorf("Error Sending message through ChatStream: %s", err)
	}
	return nil
}
func (d *Handler) start() error {
	discPeriod := viper.GetDuration("peer.discovery.period")
	tickChan := time.NewTicker(discPeriod).C
	peerLogger.Debug("Starting Peer discovery service")
	for {
		select {
		case <-tickChan:
			if err := d.SendMessage(&pb.OpenchainMessage{Type: pb.OpenchainMessage_DISC_GET_PEERS}); err != nil {
				peerLogger.Error(fmt.Sprintf("Error sending %s during handler discovery tick: %s", pb.OpenchainMessage_DISC_GET_PEERS, err))
			}
		case <-d.doneChan:
			peerLogger.Debug("Stopping discovery service")
			return nil
		}
	}
}
func (d *Handler) RequestBlocks(syncBlockRange *pb.SyncBlockRange) (<-chan *pb.SyncBlocks, error) {
	d.syncBlocksMutex.Lock()
	defer d.syncBlocksMutex.Unlock()
	if d.syncBlocks != nil {
		close(d.syncBlocks)
	}
	d.syncBlocks = make(chan *pb.SyncBlocks, viper.GetInt("peer.sync.blocks.channelSize"))
	syncBlockRangeBytes, err := proto.Marshal(syncBlockRange)
	if err != nil {
		return nil, fmt.Errorf("Error marshaling syncBlockRange during GetBlocks: %s", err)
	}
	peerLogger.Debug("Sending %s with Range %s", pb.OpenchainMessage_SYNC_GET_BLOCKS.String(), syncBlockRange)
	if err := d.SendMessage(&pb.OpenchainMessage{Type: pb.OpenchainMessage_SYNC_GET_BLOCKS, Payload: syncBlockRangeBytes}); err != nil {
		return nil, fmt.Errorf("Error sending %s during GetBlocks: %s", pb.OpenchainMessage_SYNC_GET_BLOCKS, err)
	}
	return d.syncBlocks, nil
}
func (d *Handler) beforeSyncGetBlocks(e *fsm.Event) {
	peerLogger.Debug("Received message: %s", e.Event)
	msg, ok := e.Args[0].(*pb.OpenchainMessage)
	if !ok {
		e.Cancel(fmt.Errorf("Received unexpected message type"))
		return
	}
	syncBlockRange := &pb.SyncBlockRange{}
	err := proto.Unmarshal(msg.Payload, syncBlockRange)
	if err != nil {
		e.Cancel(fmt.Errorf("Error unmarshalling SyncBlockRange in GetBlocks: %s", err))
		return
	}
	go d.sendBlocks(syncBlockRange)
}
func (d *Handler) beforeSyncBlocks(e *fsm.Event) {
	peerLogger.Debug("Received message: %s", e.Event)
	msg, ok := e.Args[0].(*pb.OpenchainMessage)
	if !ok {
		e.Cancel(fmt.Errorf("Received unexpected message type"))
		return
	}
	syncBlocks := &pb.SyncBlocks{}
	err := proto.Unmarshal(msg.Payload, syncBlocks)
	if err != nil {
		e.Cancel(fmt.Errorf("Error unmarshalling SyncBlocks in beforeSyncBlocks: %s", err))
		return
	}
	peerLogger.Debug("TODO: send received syncBlocks for start = %d and end = %d message to channel", syncBlocks.Range.Start, syncBlocks.Range.End)
	defer func() {
		if x := recover(); x != nil {
			peerLogger.Error(fmt.Sprintf("Error sending syncBlocks to channel: %v", x))
		}
	}()
	select {
	case d.syncBlocks <- syncBlocks:
	default:
		peerLogger.Warning("Did NOT send SyncBlocks message to channel for range: %d - %d", syncBlocks.Range.Start, syncBlocks.Range.End)
	}
}
func (d *Handler) sendBlocks(syncBlockRange *pb.SyncBlockRange) {
	peerLogger.Debug("Sending blocks %d-%d", syncBlockRange.Start, syncBlockRange.End)
	var blockNums []uint64
	if syncBlockRange.Start > syncBlockRange.End {
		for i := syncBlockRange.Start; i >= syncBlockRange.End; i-- {
			blockNums = append(blockNums, i)
		}
	} else {
		for i := syncBlockRange.Start; i <= syncBlockRange.End; i++ {
			peerLogger.Debug("Appending to blockNums: %d", i)
			blockNums = append(blockNums, i)
		}
	}
	for _, currBlockNum := range blockNums {
		block, err := d.Coordinator.GetBlockByNumber(currBlockNum)
		if err != nil {
			peerLogger.Error(fmt.Sprintf("Error sending blockNum %d: %s", currBlockNum, err))
			break
		}
		syncBlocks := &pb.SyncBlocks{Range: &pb.SyncBlockRange{Start: currBlockNum, End: currBlockNum}, Blocks: []*pb.Block{block}}
		syncBlocksBytes, err := proto.Marshal(syncBlocks)
		if err != nil {
			peerLogger.Error(fmt.Sprintf("Error marshalling syncBlocks for BlockNum = %d: %s", currBlockNum, err))
			break
		}
		if err := d.SendMessage(&pb.OpenchainMessage{Type: pb.OpenchainMessage_SYNC_BLOCKS, Payload: syncBlocksBytes}); err != nil {
			peerLogger.Error(fmt.Sprintf("Error sending blockNum %d: %s", currBlockNum, err))
			break
		}
	}
}
func (d *Handler) RequestStateSnapshot() (<-chan *pb.SyncStateSnapshot, error) {
	d.snapshotRequestHandler.Lock()
	defer d.snapshotRequestHandler.Unlock()
	d.snapshotRequestHandler.reset()
	syncStateSnapshotRequest := d.snapshotRequestHandler.createRequest()
	syncStateSnapshotRequestBytes, err := proto.Marshal(syncStateSnapshotRequest)
	if err != nil {
		return nil, fmt.Errorf("Error marshaling syncStateSnapshotRequest during GetStateSnapshot: %s", err)
	}
	peerLogger.Debug("Sending %s with syncStateSnapshotRequest = %s", pb.OpenchainMessage_SYNC_STATE_GET_SNAPSHOT.String(), syncStateSnapshotRequest)
	if err := d.SendMessage(&pb.OpenchainMessage{Type: pb.OpenchainMessage_SYNC_STATE_GET_SNAPSHOT, Payload: syncStateSnapshotRequestBytes}); err != nil {
		return nil, fmt.Errorf("Error sending %s during GetStateSnapshot: %s", pb.OpenchainMessage_SYNC_STATE_GET_SNAPSHOT, err)
	}
	return d.snapshotRequestHandler.channel, nil
}