func (c *Collector) Rcv(m beehive.Msg, ctx beehive.RcvContext) error {
	res := m.Data().(StatResult)
	glog.V(2).Infof("Stat results: %+v", res)
	matrix := ctx.Dict(matrixDict)
	key := res.Switch.Key()
	v, err := matrix.Get(key)
	if err != nil {
		return fmt.Errorf("No such switch in matrix: %+v", res)
	}

	c.poller.query <- StatQuery{res.Switch}

	sw := v.(SwitchStats)
	stat, ok := sw[res.Flow]
	sw[res.Flow] = res.Bytes

	glog.V(2).Infof("Previous stats: %+v, Now: %+v", stat, res.Bytes)
	if !ok || res.Bytes-stat > c.delta {
		glog.Infof("Found an elephent flow: %+v, %+v, %+v", res, stat,
			ctx.Hive().ID())
		ctx.Emit(MatrixUpdate(res))
	}

	matrix.Put(key, sw)
	return nil
}

func (r *registry) ApplyConfChange(cc raftpb.ConfChange, gn raft.GroupNode) (
	err error) {

	r.m.Lock()
	defer r.m.Unlock()

	glog.V(2).Infof("%v applies conf change %#v for %v", r, cc, gn.Node)
	switch cc.Type {
	case raftpb.ConfChangeAddNode:
		if gn.Node != cc.NodeID {
			glog.Fatalf("invalid data in the config change: %v != %v", gn.Node,
				cc.NodeID)
		}
		if gn.Data != nil {
			hi := HiveInfo{
				ID:   gn.Node,
				Addr: gn.Data.(string),
			}
			r.addHive(hi)
			glog.V(2).Infof("%v adds hive %[email protected]%v", r, hi.ID, hi.Addr)
		}

	case raftpb.ConfChangeRemoveNode:
		r.delHive(cc.NodeID)
		glog.V(2).Infof("%v deletes hive %v", r, cc.NodeID)
	}
	return nil
}

func (h *hive) Start() error {
	h.status = hiveStarted
	h.registerSignals()
	h.startRaftNode()
	if err := h.listen(); err != nil {
		glog.Errorf("%v cannot start listener: %v", h, err)
		h.Stop()
		return err
	}
	if err := h.raftBarrier(); err != nil {
		glog.Fatalf("error when joining the cluster: %v", err)
	}
	glog.V(2).Infof("%v is in sync with the cluster", h)
	h.startQees()
	h.reloadState()

	glog.V(2).Infof("%v starts message loop", h)
	dataCh := h.dataCh.out()
	for h.status == hiveStarted {
		select {
		case m := <-dataCh:
			h.handleMsg(m.msg)

		case cmd := <-h.ctrlCh:
			h.handleCmd(cmd)
		}
	}
	return nil
}

func (of *of10Driver) handlePacketIn(in of10.PacketIn, c *ofConn) error {
	inPort := in.InPort()
	// Ignore packet-ins on switch specific ports.
	if inPort > uint16(of10.PP_MAX) {
		glog.V(2).Infof("ignoring packet-in on %v", inPort)
		return nil
	}

	port, ok := of.ofPorts[inPort]
	if !ok {
		return fmt.Errorf("of10Driver: port not found %v", inPort)
	}

	glog.V(2).Infof("packet received: %v", in)

	nomIn := nom.PacketIn{
		Node:     c.node.UID(),
		InPort:   port.UID(),
		BufferID: nom.PacketBufferID(in.BufferId()),
	}
	nomIn.Packet = nom.Packet(in.Data())
	c.ctx.Emit(nomIn)

	return nil
}

func (b *bee) ProcessStatusChange(sch interface{}) {
	switch ev := sch.(type) {
	case raft.LeaderChanged:
		glog.V(2).Infof("%v recevies leader changed event %#v", b, ev)
		if ev.New == Nil {
			// TODO(soheil): when we switch to nil during a campaign, shouldn't we
			// just change the colony?
			return
		}

		oldc := b.colony()
		oldi, err := b.hive.bee(oldc.Leader)
		if err != nil {
			glog.Fatalf("%v cannot find leader: %v", b, err)
		}
		if oldi.Hive == ev.New {
			glog.V(2).Infof("%v has no need to change %v", b, oldc)
			return
		}

		newc := oldc.DeepCopy()
		if oldc.Leader != Nil {
			newc.Leader = Nil
			newc.AddFollower(oldc.Leader)
		}
		newi := b.fellowBeeOnHive(ev.New)
		newc.DelFollower(newi.ID)
		newc.Leader = newi.ID
		b.setColony(newc)

		go b.processCmd(cmdRefreshRole{})

		if ev.New != b.hive.ID() {
			return
		}

		b.setTerm(ev.Term)

		go func() {
			// FIXME(): add raft term to make sure it's versioned.
			glog.V(2).Infof("%v is the new leader of %v", b, oldc)
			up := updateColony{
				Term: ev.Term,
				Old:  oldc,
				New:  newc,
			}

			// TODO(soheil): should we have a max retry?
			_, err := b.hive.node.ProposeRetry(hiveGroup, up,
				b.hive.config.RaftElectTimeout(), -1)
			if err != nil {
				glog.Errorf("%v cannot update its colony: %v", b, err)
			}
		}()
		// TODO(soheil): add health checks here and recruit if needed.
	}
}

func (b *bee) CommitTx() error {
	// No need to replicate and/or persist the transaction.
	if !b.app.persistent() || b.detached {
		glog.V(2).Infof("%v commits in memory transaction", b)
		b.commitTxBothLayers()
		return nil
	}

	glog.V(2).Infof("%v commits persistent transaction", b)
	return b.replicate()
}

func doHandleEchoRequest(req of.Header, res of.Header, c *ofConn) error {
	glog.V(2).Infof("Received echo request from %v", c.node)
	res.SetXid(req.Xid())
	err := c.WriteHeaders([]of.Header{res})
	if err != nil {
		return err
	}
	c.Flush()
	glog.V(2).Infof("Sent echo reply to %v", c.node)
	return nil
}

func (of *of10Driver) handlePacketIn(in of10.PacketIn, c *ofConn) error {
	inPort := in.InPort()
	// Ignore packet-ins on switch specific ports.
	if inPort > uint16(of10.PP_MAX) {
		glog.V(2).Infof("ignoring packet-in on %v", inPort)
		return nil
	}

	port, ok := of.ofPorts[inPort]
	if !ok {
		return fmt.Errorf("of10Driver: port not found %v", inPort)
	}

	if glog.V(2) {
		glog.Infof("packet received: %v", in)
	}

	nomIn := nom.PacketIn{
		Node:     c.node.UID(),
		InPort:   port.UID(),
		BufferID: nom.PacketBufferID(in.BufferId()),
	}
	nomIn.Packet = nom.Packet(in.Data())
	c.ctx.Emit(nomIn)

	//c.ctx.Emit(in)

	//buf := make([]byte, 32)
	//out := of10.NewPacketOutWithBuf(buf)
	//out.Init()
	//out.SetBufferId(in.BufferId())
	//out.SetInPort(in.InPort())

	//bcast := of10.NewActionOutput()
	//bcast.SetPort(uint16(of10.PP_FLOOD))

	//out.AddActions(bcast.ActionHeader)

	//if in.BufferId() == 0xFFFFFFFF {
	//for _, d := range in.Data() {
	//out.AddData(d)
	//}
	//} else {
	//out.SetBufferId(in.BufferId())
	//}

	//c.wCh <- out.Header
	//if err := c.WriteHeader(out.Header); err != nil {
	//return fmt.Errorf("Error in writing a packet out: %v", err)
	//}
	return nil
}

func (b *bee) replicate() error {
	glog.V(2).Infof("%v replicates transaction", b)
	b.Lock()

	if b.stateL2 != nil {
		err := b.commitTxL2()
		b.stateL2 = nil
		if err != nil && err != state.ErrNoTx {
			b.Unlock()
			return err
		}
	}

	if b.stateL1.TxStatus() != state.TxOpen {
		b.Unlock()
		return state.ErrNoTx
	}

	stx := b.stateL1.Tx()
	if len(stx.Ops) == 0 {
		err := b.commitTxL1()
		b.Unlock()
		return err
	}

	b.Unlock()

	if err := b.maybeRecruitFollowers(); err != nil {
		return err
	}

	msgs := make([]*msg, len(b.msgBufL1))
	copy(msgs, b.msgBufL1)
	tx := tx{
		Tx:   stx,
		Msgs: msgs,
	}
	ctx, cnl := context.WithTimeout(context.Background(),
		10*b.hive.config.RaftElectTimeout())
	defer cnl()
	commit := commitTx{
		Tx:   tx,
		Term: b.term(),
	}
	if _, err := b.hive.node.Propose(ctx, b.group(), commit); err != nil {
		glog.Errorf("%v cannot replicate the transaction: %v", b, err)
		return err
	}
	glog.V(2).Infof("%v successfully replicates transaction", b)
	return nil
}

func (h *arpPktInHandler) Rcv(msg bh.Msg, ctx bh.RcvContext) error {
	pin := msg.Data().(nom.PacketIn)
	p := gopacket.NewPacket([]byte(pin.Packet), layers.LayerTypeEthernet, gopacket.Default)
	etherlayer := p.Layer(layers.LayerTypeEthernet)

	if etherlayer == nil {
		return nil
	}

	e, _ := etherlayer.(*layers.Ethernet)

	if e.EthernetType != layers.EthernetTypeARP {
		return nil
	}

	host, _, err := decodeARP([]byte(pin.Packet))
	host.Node = pin.Node

	if err != nil {
		glog.Errorf("ARP decoding error: %v", err)
		return err
	}
	glog.V(2).Infof("Host detected: %v", host)

	ctx.Emit(nom.HostConnected(host))

	return nil
}

func (q *qee) handleUnicastMsg(mh msgAndHandler) {
	glog.V(2).Infof("unicast msg: %v", mh.msg)
	b, ok := q.beeByID(mh.msg.To())
	if !ok {
		info, err := q.hive.registry.bee(mh.msg.To())
		if err != nil {
			glog.Errorf("cannot find bee %v", mh.msg.To())
		}

		if q.isLocalBee(info) {
			glog.Fatalf("%v cannot find local bee %v", q, mh.msg.To())
		}

		if b, ok = q.beeByID(info.ID); !ok {
			if b, err = q.newProxyBee(info); err != nil {
				glog.Errorf("%v cannnot find remote bee %v", q, mh.msg.To())
				return
			}
		}
	}

	if mh.handler == nil && !b.detached && !b.proxy {
		glog.Fatalf("handler is nil for message %v", mh.msg)
	}

	b.enqueMsg(mh)
}

func (b *bee) handoff(to uint64) error {
	if !b.app.persistent() {
		return b.handoffNonPersistent(to)
	}

	c := b.colony()
	if !c.IsFollower(to) {
		return fmt.Errorf("%v is not a follower of %v", to, b)
	}

	if _, err := b.qee.sendCmdToBee(to, cmdSync{}); err != nil {
		return err
	}

	ch := make(chan error)
	go func() {
		// TODO(soheil): use context with deadline here.
		_, err := b.qee.sendCmdToBee(to, cmdCampaign{})
		ch <- err
	}()

	t := b.hive.config.RaftElectTimeout()
	time.Sleep(t)
	if _, err := b.hive.node.ProposeRetry(c.ID, noOp{}, t, 10); err != nil {
		glog.Errorf("%v cannot sync raft: %v", b, err)
	}

	if b.isFollower(b.ID()) {
		glog.V(2).Infof("%v successfully handed off leadership to %v", b, to)
		b.becomeFollower()
	}
	return <-ch
}

func (r *registry) doApply(req interface{}) (interface{}, error) {
	glog.V(2).Infof("%v applies: %#v", r, req)

	switch req := req.(type) {
	case noOp:
		return nil, nil
	case newHiveID:
		return r.newHiveID(), nil
	case allocateBeeIDs:
		return r.allocBeeIDs(req)
	case addBee:
		return nil, r.addBee(BeeInfo(req))
	case delBee:
		return nil, r.delBee(uint64(req))
	case moveBee:
		return nil, r.moveBee(req)
	case updateColony:
		return nil, r.updateColony(req)
	case lockMappedCell:
		return r.lockCell(req)
	case transferCells:
		return nil, r.transfer(req)
	case batchReq:
		return r.handleBatch(req), nil
	}

	glog.Errorf("%v cannot handle %v", r, req)
	return nil, ErrUnsupportedRequest
}

func (d *Driver) Rcv(m beehive.Msg, ctx beehive.RcvContext) error {
	if m.NoReply() {
		return nil
	}

	q, ok := m.Data().(StatQuery)
	if !ok {
		return nil
	}

	s, ok := d.switches[q.Switch]
	if !ok {
		return fmt.Errorf("No switch stored in the driver: %+v", s)
	}

	for i, f := range s.Flows {
		f.Bytes += uint64(rand.Intn(maxSpike))
		s.Flows[i] = f
		glog.V(2).Infof("Emitting stat result for %+v", f)
		ctx.Emit(StatResult{q, f.Flow, f.Bytes})
	}

	d.switches[q.Switch] = s
	return nil
}

func (g *group) applyConfChange(e raftpb.Entry) error {
	var cc raftpb.ConfChange
	pbutil.MustUnmarshal(&cc, e.Data)
	glog.V(2).Infof("%v applies conf change %v: %#v", g, e.Index, cc)

	if len(cc.Context) == 0 {
		g.stateMachine.ApplyConfChange(cc, GroupNode{})
		return nil
	}

	if id, req, err := g.node.decReq(cc.Context); err == nil {
		if gn, ok := req.Data.(GroupNode); ok {
			res := Response{ID: id}
			res.Err = g.stateMachine.ApplyConfChange(cc, gn)
			g.node.line.call(res)
			return nil
		}
	}

	var gn GroupNode
	if err := bhgob.Decode(&gn, cc.Context); err != nil {
		glog.Fatalf("%v cannot decode config change: %v", g, err)
	}

	if gn.Node != cc.NodeID {
		glog.Fatalf("invalid config change: %v != %v", gn.Node, cc.NodeID)
	}
	g.stateMachine.ApplyConfChange(cc, gn)
	return nil
}

func (h *hive) stopQees() {
	glog.Infof("%v is stopping qees...", h)
	qs := make(map[*qee]bool)
	for _, mhs := range h.qees {
		for _, mh := range mhs {
			qs[mh.q] = true
		}
	}

	stopCh := make(chan cmdResult)
	for q := range qs {
		q.ctrlCh <- newCmdAndChannel(cmdStop{}, h.ID(), q.app.Name(), 0, stopCh)
		glog.V(3).Infof("waiting on a qee: %v", q)
		stopped := false
		tries := 5
		for !stopped {
			select {
			case res := <-stopCh:
				_, err := res.get()
				if err != nil {
					glog.Errorf("error in stopping a qee: %v", err)
				}
				stopped = true
			case <-time.After(1 * time.Second):
				if tries--; tries < 0 {
					glog.Infof("giving up on qee %v", q)
					stopped = true
					continue
				}
				glog.Infof("still waiting for a qee %v...", q)
			}
		}
	}
}

func (b *bee) createGroup() error {
	c := b.colony()
	if c.IsNil() || c.ID == Nil {
		return fmt.Errorf("%v is in no colony", b)
	}
	cfg := raft.GroupConfig{
		ID:             c.ID,
		Name:           b.String(),
		StateMachine:   b,
		Peers:          b.peers(),
		DataDir:        b.statePath(),
		SnapCount:      1024,
		FsyncTick:      b.hive.config.RaftFsyncTick,
		ElectionTicks:  b.hive.config.RaftElectTicks,
		HeartbeatTicks: b.hive.config.RaftHBTicks,
		MaxInFlights:   b.hive.config.RaftInFlights,
		MaxMsgSize:     b.hive.config.RaftMaxMsgSize,
	}
	if err := b.hive.node.CreateGroup(context.TODO(), cfg); err != nil {
		return err
	}

	if err := b.raftBarrier(); err != nil {
		return err
	}

	b.enableEmit()
	glog.V(2).Infof("%v started its raft node", b)
	return nil
}

func (h *newLinkHandler) Rcv(msg bh.Msg, ctx bh.RcvContext) error {
	l := nom.Link(msg.Data().(NewLink))
	n, _ := nom.ParsePortUID(l.From)
	d := ctx.Dict(nodeDict)
	k := string(n)
	v, err := d.Get(k)
	if err != nil {
		return nil
	}

	np := v.(nodePortsAndLinks)

	if oldl, ok := np.linkFrom(l.From); ok {
		if oldl.UID() == l.UID() {
			return nil
		}
		np.removeLink(oldl)
		ctx.Emit(nom.LinkDeleted(oldl))
	}

	glog.V(2).Infof("Link detected %v", l)
	ctx.Emit(nom.LinkAdded(l))
	np.L = append(np.L, l)
	return d.Put(k, np)
}

func peersInfo(addrs []string) map[uint64]HiveInfo {
	if len(addrs) == 0 {
		return nil
	}

	ch := make(chan []HiveInfo, len(addrs))
	for _, a := range addrs {
		go func(a string) {
			s, err := getHiveState(a)
			if err != nil {
				glog.Errorf("cannot communicate with %v: %v", a, err)
				return
			}
			ch <- s.Peers
		}(a)
	}

	// Return the first one.
	hives := <-ch
	glog.V(2).Infof("found live hives: %v", hives)
	infos := make(map[uint64]HiveInfo)
	for _, h := range hives {
		infos[h.ID] = h
	}
	return infos
}

func (b *bee) Apply(req interface{}) (interface{}, error) {
	b.Lock()
	defer b.Unlock()

	switch r := req.(type) {
	case commitTx:
		if b.txTerm < r.Term {
			b.txTerm = r.Term
		} else if r.Term < b.txTerm {
			return nil, ErrOldTx
		}

		glog.V(2).Infof("%v commits %v", b, r)
		leader := b.isLeader()

		if b.stateL2 != nil {
			b.stateL2 = nil
			glog.Errorf("%v has an L2 transaction", b)
		}

		if b.stateL1.TxStatus() == state.TxOpen {
			if !leader {
				glog.Errorf("%v is a follower and has an open transaction", b)
			}
			b.resetTx(b.stateL1, &b.msgBufL1)
		}

		if err := b.stateL1.Apply(r.Tx.Ops); err != nil {
			return nil, err
		}

		if leader && b.emitInRaft {
			for _, msg := range r.Tx.Msgs {
				msg.MsgFrom = b.beeID
				glog.V(2).Infof("%v emits %#v", b, msg)
			}
			b.throttle(r.Tx.Msgs)
		}
		return nil, nil

	case noOp:
		return nil, nil
	}
	glog.Errorf("%v cannot handle %v", b, req)
	return nil, ErrUnsupportedRequest
}