func (m *Memberlist) udpHandler() {
	for {
		select {
		case msg := <-m.handoff:
			msgType := msg.msgType
			buf := msg.buf
			from := msg.from
			switch msgType {
			case suspectMsg:
				m.handleSuspect(buf, from)
			case aliveMsg:
				m.handleAlive(buf, from)
			case deadMsg:
				m.handleDead(buf, from)
			case userMsg:
				m.handleUser(buf, from)
			default:
				m.logger.Printf("[ERR] memberlist: UDP msg type (%d) not supported. From: %s (handler)", msgType, from)
			}
		case <-m.shutdownCh:
			return
		}
	}
}
udpHandler processes messages received over UDP, but is decoupled
from the listener to avoid blocking the listener which may cause
ping/ack messages to be delayed.
package memberlist
import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"net"
	"time"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/armon/go-metrics"
	"github.com/socketplane/socketplane/Godeps/_workspace/src/github.com/hashicorp/go-msgpack/codec"
)
const (
	ProtocolVersionMin uint8 = 1
	ProtocolVersionMax       = 2
)
type messageType uint8
const (
	pingMsg messageType = iota
	indirectPingMsg
	ackRespMsg
	suspectMsg
	aliveMsg
	deadMsg
	pushPullMsg
	compoundMsg
	userMsg  
	compressMsg
	encryptMsg
)
type compressionType uint8
const (
	lzwAlgo compressionType = iota
)
const (
	MetaMaxSize            = 512  
	compoundHeaderOverhead = 2    
	compoundOverhead       = 2    
	udpBufSize             = 65536
	udpRecvBuf             = 2 * 1024 * 1024
	udpSendBuf             = 1400
	userMsgOverhead        = 1
	blockingWarning        = 10 * time.Millisecond  
	maxPushStateBytes      = 10 * 1024 * 1024
)
type ping struct {
	SeqNo uint32
	Node string
}
type indirectPingReq struct {
	SeqNo  uint32
	Target []byte
	Port   uint16
	Node   string
}
type ackResp struct {
	SeqNo uint32
}
type suspect struct {
	Incarnation uint32
	Node        string
	From        string  
}
type alive struct {
	Incarnation uint32
	Node        string
	Addr        []byte
	Port        uint16
	Meta        []byte
	Vsn []uint8
}
type dead struct {
	Incarnation uint32
	Node        string
	From        string  
}
type pushPullHeader struct {
	Nodes        int
	UserStateLen int   
	Join         bool  
}
type pushNodeState struct {
	Name        string
	Addr        []byte
	Port        uint16
	Meta        []byte
	Incarnation uint32
	State       nodeStateType
	Vsn         []uint8  
}
type compress struct {
	Algo compressionType
	Buf  []byte
}
type msgHandoff struct {
	msgType messageType
	buf     []byte
	from    net.Addr
}
func (m *Memberlist) encryptionVersion() encryptionVersion {
	switch m.ProtocolVersion() {
	case 1:
		return 0
	default:
		return 1
	}
}
func setUDPRecvBuf(c *net.UDPConn) {
	size := udpRecvBuf
	for {
		if err := c.SetReadBuffer(size); err == nil {
			break
		}
		size = size / 2
	}
}
func (m *Memberlist) tcpListen() {
	for {
		conn, err := m.tcpListener.AcceptTCP()
		if err != nil {
			if m.shutdown {
				break
			}
			m.logger.Printf("[ERR] memberlist: Error accepting TCP connection: %s", err)
			continue
		}
		go m.handleConn(conn)
	}
}
func (m *Memberlist) handleConn(conn *net.TCPConn) {
	m.logger.Printf("[DEBUG] memberlist: Responding to push/pull sync with: %s", conn.RemoteAddr())
	defer conn.Close()
	metrics.IncrCounter([]string{"memberlist", "tcp", "accept"}, 1)
	join, remoteNodes, userState, err := m.readRemoteState(conn)
	if err != nil {
		m.logger.Printf("[ERR] memberlist: Failed to receive remote state: %s", err)
		return
	}
	if err := m.sendLocalState(conn, join); err != nil {
		m.logger.Printf("[ERR] memberlist: Failed to push local state: %s", err)
	}
	if err := m.verifyProtocol(remoteNodes); err != nil {
		m.logger.Printf("[ERR] memberlist: Push/pull verification failed: %s", err)
		return
	}
	if join && m.config.Merge != nil {
		nodes := make([]*Node, len(remoteNodes))
		for idx, n := range remoteNodes {
			nodes[idx] = &Node{
				Name: n.Name,
				Addr: n.Addr,
				Port: n.Port,
				Meta: n.Meta,
				PMin: n.Vsn[0],
				PMax: n.Vsn[1],
				PCur: n.Vsn[2],
				DMin: n.Vsn[3],
				DMax: n.Vsn[4],
				DCur: n.Vsn[5],
			}
		}
		if m.config.Merge.NotifyMerge(nodes) {
			m.logger.Printf("[WARN] memberlist: Cluster merge canceled")
			return
		}
	}
	m.mergeState(remoteNodes)
	if m.config.Delegate != nil {
		m.config.Delegate.MergeRemoteState(userState, join)
	}
}
func (m *Memberlist) udpListen() {
	var n int
	var addr net.Addr
	var err error
	var lastPacket time.Time
	for {
		if !lastPacket.IsZero() && time.Now().Sub(lastPacket) > blockingWarning {
			diff := time.Now().Sub(lastPacket)
			m.logger.Printf(
				"[DEBUG] memberlist: Potential blocking operation. Last command took %v",
				diff)
		}
		buf := make([]byte, udpBufSize)
		n, addr, err = m.udpListener.ReadFrom(buf)
		if err != nil {
			if m.shutdown {
				break
			}
			m.logger.Printf("[ERR] memberlist: Error reading UDP packet: %s", err)
			continue
		}
		if n < 1 {
			m.logger.Printf("[ERR] memberlist: UDP packet too short (%d bytes). From: %s",
				len(buf), addr)
			continue
		}
		lastPacket = time.Now()
		metrics.IncrCounter([]string{"memberlist", "udp", "received"}, float32(n))
		m.ingestPacket(buf[:n], addr)
	}
}
func (m *Memberlist) ingestPacket(buf []byte, from net.Addr) {
	if m.config.EncryptionEnabled() {
		plain, err := decryptPayload(m.config.Keyring.GetKeys(), buf, nil)
		if err != nil {
			m.logger.Printf("[ERR] memberlist: Decrypt packet failed: %v", err)
			return
		}
		buf = plain
	}
	m.handleCommand(buf, from)
}
func (m *Memberlist) handleCommand(buf []byte, from net.Addr) {
	msgType := messageType(buf[0])
	buf = buf[1:]
	switch msgType {
	case compoundMsg:
		m.handleCompound(buf, from)
	case compressMsg:
		m.handleCompressed(buf, from)
	case pingMsg:
		m.handlePing(buf, from)
	case indirectPingMsg:
		m.handleIndirectPing(buf, from)
	case ackRespMsg:
		m.handleAck(buf, from)
	case suspectMsg:
		fallthrough
	case aliveMsg:
		fallthrough
	case deadMsg:
		fallthrough
	case userMsg:
		select {
		case m.handoff <- msgHandoff{msgType, buf, from}:
		default:
			m.logger.Printf("[WARN] memberlist: UDP handler queue full, dropping message (%d)", msgType)
		}
	default:
		m.logger.Printf("[ERR] memberlist: UDP msg type (%d) not supported. From: %s", msgType, from)
	}
}