// Must be called from the processing goroutine.
func processNack(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Nack
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	// If we don't consider ourselves the leader, discard.
	if !amLeader {
		return
	}

	msgProposal, msgLeader := *msg.Proposal, uint16(*msg.Leader)
	proposal, leader := store.Proposal()
	if msgProposal == proposal && msgLeader == leader {
		return
	}
	if store.CompareProposals(msgProposal, msgLeader, proposal, leader) {
		stopBeingLeader()
		store.SetProposal(msgProposal, msgLeader)
	}
}

// Must be called from the processing goroutine.
func processAccept(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Accept
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	proposal, leader := store.Proposal()
	msgProposal, msgLeader := *msg.Proposal, node
	if proposal != msgProposal || leader != msgLeader {
		// Send a nack message and return,
		// if this accept relates to an earlier proposal.
		if store.CompareProposals(proposal, leader, msgProposal,
			msgLeader) {

			var nack coproto.Nack
			nack.Proposal = new(uint64)
			nack.Leader = new(uint32)
			*nack.Proposal = proposal
			*nack.Leader = uint32(leader)
			conn.SendProto(6, &nack)

			return
		}

		store.SetProposal(msgProposal, msgLeader)
	}

	addAccept(node, &msg)
}

// Must be called from the processing goroutine.
func processPromise(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Promise
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	if receivedPromises == nil {
		// Not attempting to become leader.
		log.Print("core/consensus: discarded promise, not becoming "+
			"leader, from ", node)
		return
	}

	proposal, leader := store.Proposal()
	if proposal != *msg.Proposal || leader != uint16(*msg.Leader) {
		log.Print("core/consensus: rejected promise for wrong "+
			"proposal number from ", node)
		return
	}
	if receivedPromises[node] != nil {
		// Protocol violation; shouldn't get duplicate promises.
		log.Print("core/consensus: PROTOCOL VIOLATION: received "+
			"duplicate promise from node ", node)
		return
	}

	log.Print("core/consensus: received promise from node ", node)
	addPromise(node, &msg)
}

// Must be called from the processing goroutine.
func processAccepted(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Accepted
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	addAccepted(node, &msg)
}

func handleConn(node uint16, conn *connect.BaseConn) {

	for {
		msg, ok := <-conn.Received
		if !ok {
			break
		}

		// We only have one message type.
		if *msg.MsgType != 2 {
			conn.Close()
		}

		handleForward(conn, msg.Content)
	}

	terminatedConnCh <- receivedConn{node: node, conn: conn}
}

func handleForward(conn *connect.BaseConn, content []byte) {
	var msg rproto.Forward
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	userMsg := new(UserMessage)
	userMsg.Sender = *msg.Sender
	userMsg.Recipient = *msg.Recipient
	userMsg.Tag = *msg.Tag
	userMsg.Content = *msg.Content
	userMsg.Ttl = uint16(*msg.Ttl)

	// Call message received callback.
	for _, cb := range receivedCallbacks {
		cb(userMsg)
	}
}

func sendMessages(conn *connect.BaseConn) {

	var sessionId uint64
	for key := range sinkUserData {
		if strings.HasPrefix(key, "attach ") {
			idStr := key[7:]
			sessionId, _ = strconv.ParseUint(idStr, 10, 64)
		}
	}

	var msg cliproto_up.Send
	msg.Recipient = new(uint64)
	msg.Tag = new(string)
	msg.Content = new(string)
	*msg.Recipient = sessionId
	*msg.Tag = "benchmark"

	i := uint64(0)
	prevSec := time.Now().Unix()
	count := uint(0)
	for {
		sec := time.Now().Unix()
		if sec != prevSec {
			prevSec = sec
			count = 0
		}

		// If we've hit our cap, busyloop until we can send again.
		if *rate != 0 && count >= *rate {
			continue
		}

		i++
		count++

		timeStr := strconv.FormatInt(time.Now().UnixNano(), 10)
		numberStr := strconv.FormatUint(i, 10)
		*msg.Content = timeStr + " " + numberStr

		conn.SendProto(6, &msg)
	}
}

func incomingConn(node uint16, conn *connect.BaseConn) {

	store.StartTransaction()
	if store.Degraded() {
		conn.Close()
		store.EndTransaction()
		return
	}

	userConn := new(userConn)
	userConn.conn = conn
	userConn.deliver = make(chan *relay.UserMessage, 100)

	// Add to connections set.
	connectionsLock.Lock()
	connections[userConn] = true
	connectionsLock.Unlock()

	store.EndTransaction()

	go handleConn(userConn)
}

func handleMsg(node uint16, conn *connect.BaseConn, msg *baseproto.Message) {
	if *msg.MsgType == 2 {
		// Change Forward message.
		forward := new(chproto.ChangeForward)
		err := proto.Unmarshal(msg.Content, forward)
		if err != nil {
			conn.Close()
		}

		// If this is not a core node,
		// and the requesting node is not this node,
		// discard the message.
		if uint16(*forward.Request.RequestNode) != config.Id() &&
			!config.IsCore() {
			return
		}

		// Send a Change Forward Ack message to the sender.
		ack := new(chproto.ChangeForwardAck)
		ack.RequestNode = forward.Request.RequestNode
		ack.RequestId = forward.Request.RequestId
		conn.SendProto(3, ack)

		// Send the forward message to our processing goroutine.
		receivedForwardCh <- receivedForward{forward: forward,
			node: node}

	} else if *msg.MsgType == 3 {
		// Change Forward Ack message.
		ack := new(chproto.ChangeForwardAck)
		err := proto.Unmarshal(msg.Content, ack)
		if err != nil {
			conn.Close()
		}

		// Send the ack message to our processing goroutine.
		receivedAckCh <- receivedAck{ack: ack, node: node}

	} else {
		// Unknown message.
		conn.Close()
	}
}

// Must be called from the processing goroutine.
func processPrepare(node uint16, conn *connect.BaseConn, content []byte) {
	var msg coproto.Prepare
	if err := proto.Unmarshal(content, &msg); err != nil {
		conn.Close()
		return
	}

	store.StartTransaction()
	defer store.EndTransaction()

	newProposal, newLeader := *msg.Proposal, node
	proposal, leader := store.Proposal()
	if store.CompareProposals(newProposal, newLeader, proposal, leader) {

		log.Print("core/consensus: sending promise to ", newLeader)

		// Create a promise message to send back.
		var promise coproto.Promise
		promise.Proposal = new(uint64)
		promise.Leader = new(uint32)
		promise.PrevProposal = new(uint64)
		promise.PrevLeader = new(uint32)
		*promise.Proposal = newProposal
		*promise.Leader = uint32(newLeader)
		*promise.PrevProposal = proposal
		*promise.PrevLeader = uint32(leader)

		// Add all the instructions we've previously accepted or chosen.
		slots := store.InstructionSlots()
		theirFirstUnapplied := *msg.FirstUnapplied
		ourStart := store.InstructionStart()
		relativeSlot := int(theirFirstUnapplied - ourStart)
		if relativeSlot < 0 {
			relativeSlot = 0
		}
		var accepted []*coproto.Instruction
		for ; relativeSlot < len(slots); relativeSlot++ {
			slot := slots[relativeSlot]
			slotNum := ourStart + uint64(relativeSlot)
			for i, _ := range slot {
				if slot[i].IsChosen() {
					appendInst(&accepted, slotNum, slot[i])
					break
				}

				weAccepted := false
				for _, node := range slot[i].Accepted() {
					if node == config.Id() {
						weAccepted = true
						break
					}
				}

				if weAccepted {
					appendInst(&accepted, slotNum, slot[i])
					break
				}
			}
		}

		// Send promise message.
		conn.SendProto(3, &promise)

		// Accept the other node as our new leader.
		store.SetProposal(newProposal, newLeader)
	} else {
		var nack coproto.Nack
		nack.Proposal = new(uint64)
		nack.Leader = new(uint32)
		*nack.Proposal = proposal
		*nack.Leader = uint32(leader)
		conn.SendProto(6, &nack)
	}
}