func (op *obcSieve) Drain() {
	op.pbft.drain()
}
Drain will block until all remaining execution has been handled
package obcpbft
import (
	"encoding/base64"
	"fmt"
	"reflect"
	"github.com/golang/protobuf/proto"
	"github.com/openblockchain/obc-peer/openchain/consensus"
	"github.com/openblockchain/obc-peer/openchain/util"
	pb "github.com/openblockchain/obc-peer/protos"
	"github.com/spf13/viper"
)
type obcSieve struct {
	stack consensus.Stack
	pbft  *pbftCore
	id            uint64
	epoch         uint64
	imminentEpoch uint64
	blockNumber   uint64
	currentReq    string
	currentResult []byte
	currentTx   []*pb.Transaction
	verifyStore []*Verify
	queuedExec map[uint64]*Execute
	queuedTx   [][]byte
}
func newObcSieve(id uint64, config *viper.Viper, stack consensus.Stack) *obcSieve {
	op := &obcSieve{stack: stack, id: id}
	op.queuedExec = make(map[uint64]*Execute)
	op.pbft = newPbftCore(id, config, op, stack)
	op.pbft.sts.RegisterListener(op)
	return op
}
func (op *obcSieve) moreCorrectThanByzantineQuorum() int {
	return 2*op.pbft.f + 1
}
func (op *obcSieve) RecvMsg(ocMsg *pb.OpenchainMessage, senderHandle *pb.PeerID) error {
	op.pbft.lock()
	defer op.pbft.unlock()
	if ocMsg.Type == pb.OpenchainMessage_CHAIN_TRANSACTION {
		logger.Info("New consensus request received")
		op.broadcastMsg(&SieveMessage{&SieveMessage_Request{ocMsg.Payload}})
		op.recvRequest(ocMsg.Payload)
		return nil
	}
	if ocMsg.Type != pb.OpenchainMessage_CONSENSUS {
		return fmt.Errorf("Unexpected message type: %s", ocMsg.Type)
	}
	senderID, err := getValidatorID(senderHandle)
	if err != nil {
		panic("Cannot map sender's PeerID to a valid replica ID")
	}
	svMsg := &SieveMessage{}
	err = proto.Unmarshal(ocMsg.Payload, svMsg)
	if err != nil {
		err = fmt.Errorf("Could not unmarshal sieve message: %v", ocMsg)
		logger.Error(err.Error())
		return err
	}
	if req := svMsg.GetRequest(); req != nil {
		op.recvRequest(req)
	} else if exec := svMsg.GetExecute(); exec != nil {
		if senderID != exec.ReplicaId {
			logger.Warning("Sender ID included in message (%v) doesn't match ID corresponding to the receiving stream (%v)", exec.ReplicaId, senderID)
			return nil
		}
		op.recvExecute(exec)
	} else if verify := svMsg.GetVerify(); verify != nil {
		op.recvVerify(verify)
	} else if pbftMsg := svMsg.GetPbftMessage(); pbftMsg != nil {
		op.pbft.unlock()
		op.pbft.receive(pbftMsg, senderID)
		op.pbft.lock()
	} else {
		err = fmt.Errorf("Received invalid sieve message: %v", svMsg)
		logger.Error(err.Error())
	}
	return nil
}
func (op *obcSieve) Close() {
	op.pbft.close()
}