func (this *service) onPublish(msg *message.PublishMessage) error {
	if msg.Retain() {
		if err := this.topicsMgr.Retain(msg); err != nil {
			glog.Errorf("(%s) Error retaining message: %v", this.cid(), err)
		}
	}
	err := this.topicsMgr.Subscribers(msg.Topic(), msg.QoS(), &this.subs, &this.qoss)
	if err != nil {
		glog.Errorf("(%s) Error retrieving subscribers list: %v", this.cid(), err)
		return err
	}
	msg.SetRetain(false)
	for _, s := range this.subs {
		if s != nil {
			fn, ok := s.(*OnPublishFunc)
			if !ok {
				glog.Errorf("Invalid onPublish Function")
				return fmt.Errorf("Invalid onPublish Function")
			} else {
				(*fn)(msg)
			}
		}
	}
	return nil
}
onPublish() is called when the server receives a PUBLISH message AND have completed
the ack cycle. This method will get the list of subscribers based on the publish
topic, and publishes the message to the list of subscribers.
package service
import (
	"errors"
	"fmt"
	"io"
	"reflect"
	"github.com/surge/glog"
	"github.com/surgemq/message"
	"github.com/surgemq/surgemq/sessions"
)
var (
	errDisconnect = errors.New("Disconnect")
)
func (this *service) processor() {
	defer func() {
		if r := recover(); r != nil {
		}
		this.wgStopped.Done()
		this.stop()
	}()
	glog.Debugf("(%s) Starting processor", this.cid())
	this.wgStarted.Done()
	for {
		mtype, total, err := this.peekMessageSize()
		if err != nil {
			glog.Errorf("(%s) Error peeking next message size: %v", this.cid(), err)
			return
		}
		msg, n, err := this.peekMessage(mtype, total)
		if err != nil {
			glog.Errorf("(%s) Error peeking next message: %v", this.cid(), err)
			return
		}
		this.inStat.increment(int64(n))
		err = this.processIncoming(msg)
		if err != nil {
			if err != errDisconnect {
				glog.Errorf("(%s) Error processing %s: %v", this.cid(), msg.Name(), err)
			} else {
				return
			}
		}
		_, err = this.in.ReadCommit(total)
		if err != nil {
			if err != io.EOF {
				glog.Errorf("(%s) Error committing %d read bytes: %v", this.cid(), total, err)
			}
			return
		}
		if this.isDone() && this.in.Len() == 0 {
			return
		}
	}
}
func (this *service) processIncoming(msg message.Message) error {
	var err error = nil
	switch msg := msg.(type) {
	case *message.PublishMessage:
		err = this.processPublish(msg)
	case *message.PubackMessage:
		this.sess.Pub1ack.Ack(msg)
		this.processAcked(this.sess.Pub1ack)
	case *message.PubrecMessage:
		if err = this.sess.Pub2out.Ack(msg); err != nil {
			break
		}
		resp := message.NewPubrelMessage()
		resp.SetPacketId(msg.PacketId())
		_, err = this.writeMessage(resp)
	case *message.PubrelMessage:
		if err = this.sess.Pub2in.Ack(msg); err != nil {
			break
		}
		this.processAcked(this.sess.Pub2in)
		resp := message.NewPubcompMessage()
		resp.SetPacketId(msg.PacketId())
		_, err = this.writeMessage(resp)
	case *message.PubcompMessage:
		if err = this.sess.Pub2out.Ack(msg); err != nil {
			break
		}
		this.processAcked(this.sess.Pub2out)
	case *message.SubscribeMessage:
		return this.processSubscribe(msg)
	case *message.SubackMessage:
		this.sess.Suback.Ack(msg)
		this.processAcked(this.sess.Suback)
	case *message.UnsubscribeMessage:
		return this.processUnsubscribe(msg)
	case *message.UnsubackMessage:
		this.sess.Unsuback.Ack(msg)
		this.processAcked(this.sess.Unsuback)
	case *message.PingreqMessage:
		resp := message.NewPingrespMessage()
		_, err = this.writeMessage(resp)
	case *message.PingrespMessage:
		this.sess.Pingack.Ack(msg)
		this.processAcked(this.sess.Pingack)
	case *message.DisconnectMessage:
		this.sess.Cmsg.SetWillFlag(false)
		return errDisconnect
	default:
		return fmt.Errorf("(%s) invalid message type %s.", this.cid(), msg.Name())
	}
	if err != nil {
		glog.Debugf("(%s) Error processing acked message: %v", this.cid(), err)
	}
	return err
}
func (this *service) processAcked(ackq *sessions.Ackqueue) {
	for _, ackmsg := range ackq.Acked() {
		msg, err := ackmsg.Mtype.New()
		if err != nil {
			glog.Errorf("process/processAcked: Unable to creating new %s message: %v", ackmsg.Mtype, err)
			continue
		}
		if _, err := msg.Decode(ackmsg.Msgbuf); err != nil {
			glog.Errorf("process/processAcked: Unable to decode %s message: %v", ackmsg.Mtype, err)
			continue
		}
		ack, err := ackmsg.State.New()
		if err != nil {
			glog.Errorf("process/processAcked: Unable to creating new %s message: %v", ackmsg.State, err)
			continue
		}
		if _, err := ack.Decode(ackmsg.Ackbuf); err != nil {
			glog.Errorf("process/processAcked: Unable to decode %s message: %v", ackmsg.State, err)
			continue
		}
		switch ackmsg.State {
		case message.PUBREL:
			if err = this.onPublish(msg.(*message.PublishMessage)); err != nil {
				glog.Errorf("(%s) Error processing ack'ed %s message: %v", this.cid(), ackmsg.Mtype, err)
			}
		case message.PUBACK, message.PUBCOMP, message.SUBACK, message.UNSUBACK, message.PINGRESP:
			glog.Debugf("process/processAcked: %s", ack)
			err = nil
		default:
			glog.Errorf("(%s) Invalid ack message type %s.", this.cid(), ackmsg.State)
			continue
		}
		if ackmsg.OnComplete != nil {
			onComplete, ok := ackmsg.OnComplete.(OnCompleteFunc)
			if !ok {
				glog.Errorf("process/processAcked: Error type asserting onComplete function: %v", reflect.TypeOf(ackmsg.OnComplete))
			} else if onComplete != nil {
				if err := onComplete(msg, ack, nil); err != nil {
					glog.Errorf("process/processAcked: Error running onComplete(): %v", err)
				}
			}
		}
	}
}
func (this *service) processPublish(msg *message.PublishMessage) error {
	switch msg.QoS() {
	case message.QosExactlyOnce:
		this.sess.Pub2in.Wait(msg, nil)
		resp := message.NewPubrecMessage()
		resp.SetPacketId(msg.PacketId())
		_, err := this.writeMessage(resp)
		return err
	case message.QosAtLeastOnce:
		resp := message.NewPubackMessage()
		resp.SetPacketId(msg.PacketId())
		if _, err := this.writeMessage(resp); err != nil {
			return err
		}
		return this.onPublish(msg)
	case message.QosAtMostOnce:
		return this.onPublish(msg)
	}
	return fmt.Errorf("(%s) invalid message QoS %d.", this.cid(), msg.QoS())
}
func (this *service) processSubscribe(msg *message.SubscribeMessage) error {
	resp := message.NewSubackMessage()
	resp.SetPacketId(msg.PacketId())
	var retcodes []byte
	topics := msg.Topics()
	qos := msg.Qos()
	this.rmsgs = this.rmsgs[0:0]
	for i, t := range topics {
		rqos, err := this.topicsMgr.Subscribe(t, qos[i], &this.onpub)
		if err != nil {
			return err
		}
		this.sess.AddTopic(string(t), qos[i])
		retcodes = append(retcodes, rqos)
		this.topicsMgr.Retained(t, &this.rmsgs)
		glog.Debugf("(%s) topic = %s, retained count = %d", this.cid(), string(t), len(this.rmsgs))
	}
	if err := resp.AddReturnCodes(retcodes); err != nil {
		return err
	}
	if _, err := this.writeMessage(resp); err != nil {
		return err
	}
	for _, rm := range this.rmsgs {
		if err := this.publish(rm, nil); err != nil {
			glog.Errorf("service/processSubscribe: Error publishing retained message: %v", err)
			return err
		}
	}
	return nil
}
func (this *service) processUnsubscribe(msg *message.UnsubscribeMessage) error {
	topics := msg.Topics()
	for _, t := range topics {
		this.topicsMgr.Unsubscribe(t, &this.onpub)
		this.sess.RemoveTopic(string(t))
	}
	resp := message.NewUnsubackMessage()
	resp.SetPacketId(msg.PacketId())
	_, err := this.writeMessage(resp)
	return err
}