func (op *obcBatch) 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")
		if (op.pbft.primary(op.pbft.view) == op.pbft.id) && op.pbft.activeView {  
			err := op.leaderProcReq(ocMsg.Payload)
			if err != nil {
				return err
			}
		} else {  
			batchMsg := &BatchMessage{&BatchMessage_Request{ocMsg.Payload}}
			packedBatchMsg, _ := proto.Marshal(batchMsg)
			ocMsg := &pb.OpenchainMessage{
				Type:    pb.OpenchainMessage_CONSENSUS,
				Payload: packedBatchMsg,
			}
			op.stack.Broadcast(ocMsg, pb.PeerEndpoint_UNDEFINED)
		}
		return nil
	}
	if ocMsg.Type != pb.OpenchainMessage_CONSENSUS {
		return fmt.Errorf("Unexpected message type: %s", ocMsg.Type)
	}
	batchMsg := &BatchMessage{}
	err := proto.Unmarshal(ocMsg.Payload, batchMsg)
	if err != nil {
		return err
	}
	if req := batchMsg.GetRequest(); req != nil {
		if (op.pbft.primary(op.pbft.view) == op.pbft.id) && op.pbft.activeView {
			err := op.leaderProcReq(req)
			if err != nil {
				return err
			}
		}
	} else if pbftMsg := batchMsg.GetPbftMessage(); pbftMsg != nil {
		senderID, err := getValidatorID(senderHandle)  
		if err != nil {
			panic("Cannot map sender's PeerID to a valid replica ID")
		}
		op.pbft.unlock()
		op.pbft.receive(pbftMsg, senderID)
		op.pbft.lock()
	} else {
		err = fmt.Errorf("Unknown request: %+v", req)
		logger.Error(err.Error())
	}
	return nil
}
RecvMsg receives both CHAIN_TRANSACTION and CONSENSUS messages from
the stack. New transaction requests are broadcast to all replicas,
so that the current primary will receive the request.
package obcpbft
import (
	"encoding/base64"
	"fmt"
	"time"
	"github.com/openblockchain/obc-peer/openchain/consensus"
	"github.com/openblockchain/obc-peer/openchain/util"
	pb "github.com/openblockchain/obc-peer/protos"
	"github.com/golang/protobuf/proto"
	"github.com/spf13/viper"
)
type obcBatch struct {
	stack consensus.Stack
	pbft  *pbftCore
	batchSize        int
	batchStore       [][]byte
	batchTimer       *time.Timer
	batchTimerActive bool
	batchTimeout     time.Duration
}
func newObcBatch(id uint64, config *viper.Viper, stack consensus.Stack) *obcBatch {
	var err error
	op := &obcBatch{stack: stack}
	op.pbft = newPbftCore(id, config, op, stack)
	op.batchSize = config.GetInt("general.batchSize")
	op.batchStore = nil
	op.batchTimeout, err = time.ParseDuration(config.GetString("general.timeout.batch"))
	if err != nil {
		panic(fmt.Errorf("Cannot parse batch timeout: %s", err))
	}
	op.batchTimer = time.NewTimer(100 * time.Hour)  
	op.batchTimer.Stop()
	go op.batchTimerHander()
	return op
}