//
// Synchronize with the leader.
//
func syncWithPeer(pipe *common.PeerPipe,
	handler ActionHandler,
	factory MsgFactory,
	killch <-chan bool) (success bool, isKilled bool) {

	log.Current.Debugf("WatcherServer.syncWithPeer(): Watcher start synchronization with peer (TCP %s)", pipe.GetAddr())
	proxy := NewFollowerSyncProxy(pipe, handler, factory, false)
	donech := proxy.GetDoneChannel()
	go proxy.Start()
	defer proxy.Terminate()

	// This will block until NewWatcherSyncProxy has sychronized with the peer (a bool is pushed to donech)
	select {
	case success = <-donech:
		if success {
			log.Current.Debugf("WatcherServer.syncWithPeer(): Watcher done synchronization with peer (TCP %s)", pipe.GetAddr())
		}
		return success, false
	case <-killch:
		// simply return. The pipe will eventually be closed and
		// cause WatcherSyncProxy to err out.
		log.Current.Debugf("WatcherServer.syncWithPeer(): Recieve kill singal.  Synchronization with peer (TCP %s) terminated.",
			pipe.GetAddr())
		return false, true
	}
}

func send(packet common.Packet, pipe *common.PeerPipe) error {

	log.Current.Tracef("SyncProxy.send(): sending packet %s to peer (TCP %s)", packet.Name(), pipe.GetAddr())
	if !pipe.Send(packet) {
		return common.NewError(common.SERVER_ERROR, fmt.Sprintf("SyncProxy.listen(): Fail to send packet %s to peer (TCP %s)",
			packet.Name(), pipe.GetAddr()))
	}

	return nil
}

func listen(name string, pipe *common.PeerPipe) (common.Packet, error) {

	reqch := pipe.ReceiveChannel()
	req, ok := <-reqch
	if !ok {
		return nil, common.NewError(common.SERVER_ERROR, "SyncProxy.listen(): channel closed. Terminate")
	}

	if req.Name() != name {
		return nil, common.NewError(common.PROTOCOL_ERROR,
			"SyncProxy.listen(): Expect message "+name+", Receive message "+req.Name())
	}

	return req, nil
}

//
// Add a follower and starts a listener for the follower.
// If the leader is terminated, the pipe between leader
// and follower will also be closed.
//
func (l *Leader) AddFollower(fid string,
	peer *common.PeerPipe,
	o *observer) {
	l.mutex.Lock()
	defer l.mutex.Unlock()

	// AddFollower requires holding the mutex such that the leader thread
	// will not be sending new proposal or commit (see sendProposal() and
	// sendCommit()) to followers.  This allow this function to copy the
	// proposals and commits from the observer queue into the pipe, before
	// the leader has a chance to send new messages.
	for packet := o.getNext(); packet != nil; packet = o.getNext() {

		switch request := packet.(type) {
		case ProposalMsg:
			txid := common.Txnid(request.GetTxnid())
			log.Current.Debugf("Leader.AddFollower() : send observer's packet %s, txid %d", packet.Name(), txid)
		case CommitMsg:
			txid := common.Txnid(request.GetTxnid())
			log.Current.Debugf("Leader.AddFollower() : send observer's packet %s, txid %d", packet.Name(), txid)
		}

		peer.Send(packet)
	}

	// Rememeber the old message listener and start a new one.
	oldListener, ok := l.followers[fid]

	listener := newListener(fid, peer, l)
	l.followers[fid] = listener
	go listener.start()

	// kill the old message listener
	if ok && oldListener != nil {
		log.Current.Debugf("Leader.AddFollower() : old Listener found for follower %s.  Terminating old listener", fid)
		oldListener.terminate()
	} else {
		// notify a brand new follower (not just replacing an existing one)
		l.changech <- true
	}
}

//
// Synchronize with the leader.
//
func syncWithLeader(naddr string,
	pipe *common.PeerPipe,
	handler ActionHandler,
	factory MsgFactory,
	killch <-chan bool) bool {

	log.Current.Debugf("FollowerServer.syncWithLeader(): Follower %s start synchronization with leader (TCP %s)",
		naddr, pipe.GetAddr())
	proxy := NewFollowerSyncProxy(pipe, handler, factory, true)
	donech := proxy.GetDoneChannel()
	go proxy.Start()
	defer proxy.Terminate()

	// This will block until NewFollowerSyncProxy has sychronized with the leader (a bool is pushed to donech)
	select {
	case success := <-donech:
		if success {
			log.Current.Debugf("FollowerServer.syncWithLeader(): Follower %s done synchronization with leader (TCP %s)",
				naddr, pipe.GetAddr())
		}
		return success
	case <-killch:
		// simply return. The pipe will eventually be closed and
		// cause FollowerSyncProxy to err out.
		log.Current.Debugf("FollowerServer.syncWithLeader(): Recieve kill singal.  Synchronization with leader (TCP %s) terminated.",
			pipe.GetAddr())
	}

	return false
}

//
// Start a LeaderSyncProxy to synchornize the leader
// and follower state.
//
func (l *LeaderServer) startProxy(peer *common.PeerPipe) {

	defer func() {
		if r := recover(); r != nil {
			log.Current.Errorf("panic in LeaderServer.startProxy() : %s\n", r)
			log.Current.Errorf("%s", log.Current.StackTrace())
		} else {
			log.Current.Debugf("LeaderServer.startProxy() : Terminates.")
			log.Current.Tracef(log.Current.StackTrace())
		}

		// deregister the proxy with the leader Server upon exit
		l.deregisterOutstandingProxy(peer.GetAddr())
	}()

	// create a proxy that will sycnhronize with the peer.
	log.Current.Debugf("LeaderServer.startProxy(): Start synchronization with follower. Peer TCP connection (%s)", peer.GetAddr())
	proxy := NewLeaderSyncProxy(l.leader, l.consentState, peer, l.handler, l.factory)
	donech := proxy.GetDoneChannel()

	// Create an observer for the leader.  The leader will put on-going proposal msg and commit msg
	// onto the observer queue.  This ensure that we can won't miss those mutations as the leader is
	// sync'ign withe follower.  The messages in observer queue will eventually route to follower.
	o := NewObserver()
	l.leader.AddObserver(peer.GetAddr(), o)
	defer l.leader.RemoveObserver(peer.GetAddr())

	// start the proxy
	go proxy.Start(o)
	defer proxy.Terminate()

	// Get the killch for this go-routine
	killch := l.getProxyKillChan(peer.GetAddr())
	if killch == nil {
		log.Current.Debugf("LeaderServer.startProxy(): Cannot find killch for proxy (TCP connection = %s).", peer.GetAddr())
		log.Current.Debugf("LeaderServer.startProxy(): Cannot start follower sync.")
		return
	}

	// this go-routine will be blocked until handshake is completed between the
	// leader and the follower.  By then, the leader will also get majority
	// confirmation that it is a leader.
	select {
	case success := <-donech:
		if success {
			// tell the leader to add this follower for processing request.  If there is a follower running already,
			// AddFollower() will terminate the existing follower instance, and then create a new one.
			fid := proxy.GetFid()
			if proxy.CanFollowerVote() {
				l.leader.AddFollower(fid, peer, o)
				log.Current.Debugf("LeaderServer.startProxy(): Synchronization with follower %s done (TCP conn = %s).  Add follower.",
					fid, peer.GetAddr())

				// At this point, the follower has voted this server as the leader.
				// Notify the request processor to start processing new request for this host
				l.notifyReady()
			} else {
				l.leader.AddWatcher(fid, peer, o)
				log.Current.Debugf("LeaderServer.startProxy(): Sync with watcher done.  Add Watcher %s (TCP conn = %s)",
					fid, peer.GetAddr())
			}
		} else {
			log.Current.Errorf("LeaderServer:startProxy(): Leader Fail to synchronization with follower (TCP conn = %s)", peer.GetAddr())
		}
	case <-killch:
		log.Current.Infof("LeaderServer:startProxy(): Sync proxy is killed while synchronizing with follower (TCP conn == %s)",
			peer.GetAddr())
	}
}