命令记不住

pushserver代码详解

1. push server

2.启动所有服务

启动所有服务

pushone\app\controllers\app.go

func init() {
	fmt.Printf("%s\n", os.Args[0])
	l4g.Debug("start")
	runtime.GOMAXPROCS(runtime.NumCPU())
	l4g.LoadConfiguration("../../conf/example.xml")
	go configsvr.Start("../../conf/myappconfig.json")
	time.Sleep(200)
	l4g.Debug("Home: " + configsvr.Fetch("home"))
	//l4g.LoadConfiguration(configsvr.Fetch("home") + "../conf/example.xml")

	l4g.Debug("channelsvr.Start")
	go channelsvr.Start()
	time.Sleep(10)
	l4g.Debug("channelsvr.StartTcpSvr")
	go channelsvr.StartTcpSvr()
	time.Sleep(10)
	l4g.Debug("sockstore.Start")
	go sockstore.Start()
	l4g.Debug("msgstore.Start")
	go msgstore.Start()
	l4g.Debug("channelsvr.RpcClientStart")
	go channelsvr.RpcClientStart()
	l4g.Debug("channelsvr.RpcServerStart")
	go channelsvr.RpcServerStart()
	time.Sleep(200)
	l4g.Debug("StartRpcClient")
	go StartRpcClient()
}

g

g

g

3.控制中心，socket字典本地存储 msg center

控制中心，socket字典本地存储 msg center

channelsvr.Start

pushservice\channelsvr\channelsvr.go

type tChanGet struct {
	Key          string
	ReplyChannel chan *tTPCChans
}

type tChanRegister struct {
	clientId   string
	controller *tTPCChans
}

var once sync.Once

type tSocketMapChans struct {
	chGet       chan *tChanGet
	chRegister  chan *tChanRegister
	chDelete    chan string
	chBroadcast chan string
	//返回在线用户数量
	chGetCount chan chan int
}

var socketMapChans = &tSocketMapChans{
	chGet:       make(chan *tChanGet, 10),
	chRegister:  make(chan *tChanRegister, 10),
	chDelete:    make(chan string, 10),
	chBroadcast: make(chan string),
	chGetCount:  make(chan chan int, 10),
}

func Start() {
	once.Do(func() {
		l4g.Debug("channelsvr.Start()")
		//存放ClientID与Go进程通道的映射map
		data := map[string]*tTPCChans{}
		defer func() {
			close(socketMapChans.chGet)
			close(socketMapChans.chRegister)
			close(socketMapChans.chDelete)
			close(socketMapChans.chBroadcast)
		}()

		for {
			select {
			case value := <-socketMapChans.chGet:
				l4g.Debug("channelsvr.Start():chGet:", value.Key)
				value.ReplyChannel <- data[value.Key]
			case value := <-socketMapChans.chRegister:
				l4g.Debug("channelsvr.Start():chRegister:", value.clientId)
				data[value.clientId] = value.controller
			case value := <-socketMapChans.chDelete:
				l4g.Debug("channelsvr.Start():chDelete:", value)
				delete(data, value)
			case value := <-socketMapChans.chBroadcast:
				l4g.Debug("channelsvr.Start():chBroadcast:", value)
				for _, v := range data {
					v.Send("100000000", value)
				}
			case reply := <-socketMapChans.chGetCount:
				l4g.Debug("channelsvr.Start():chGetCount")
				reply <- len(data)
			}
		}
	})
}

对外api

//根据clientid获取socket channel
func GetChannel(key string) *tTPCChans {
	sendData := &tChanGet{
		Key:          key,
		ReplyChannel: make(chan *tTPCChans),
	}
	defer func() {
		close(sendData.ReplyChannel)
	}()
	socketMapChans.chGet <- sendData
	return <-sendData.ReplyChannel
}

func Login(clientId string, controller *tTPCChans) {
	l4g.Debug("channelsvr.Register()::ClientID:\"%s\"", clientId)
	socketMapChans.chRegister <- &tChanRegister{
		clientId:   clientId,
		controller: controller,
	}
	//缺少，redis find, if exist, close(tcpconn)
	sockstore.Login(clientId)
}

func Logout(clientId string) {
	l4g.Debug("channelsvr.DeleteClient()::ClientID:\"%s\"", clientId)
	socketMapChans.chDelete <- clientId
	sockstore.Logout(clientId)
}

// 推送本机socket的消息
func PushLocalMessage(clientId, msgId, msg string) {
	l4g.Debug("PushMessage():%s", msg)
	if controller := GetChannel(clientId); nil != controller {
		controller.Send(msgId, msg)
	} else {
		l4g.Error("PushMessage1():\"%s\" couldn't find socket.", clientId)
	}
}

func PushMessage(clientId, msgId, msg string) {
	hostname, isLocal := sockstore.GetHostname(clientId)
	if isLocal {
		PushLocalMessage(clientId, msgId, msg)
	} else {
		RemotePushMessage(hostname, clientId, msgId, msg)
	}
}

// 给本机每个在线用户推送消息
func Broadcast(message string) {
	l4g.Debug("Broadcast: %s", message)
	socketMapChans.chBroadcast <- message
}

// 获得本机在线人数
func GetCount() int {
	ReplyChannel := make(chan int)
	defer func() {
		close(ReplyChannel)
	}()
	socketMapChans.chGetCount <- ReplyChannel
	return <-ReplyChannel
}

Login: 从tcp server获得的连接，包装之后，注册到本机的msg center。然后，存储数据到redis（sockstore）

PushMessage发送消息逻辑：查找clientID所在服务器，如果在本机，直接发送，如果在远端，通过rpc发送，调用远端机器的“本地发送”接口

4.连接对外tcp服务tcp server

pushservice\channelsvr\tcpconn.go

const TCP_READ_TIMEOUT = 30 * time.Minute

type TChanSend struct {
	MsgId   string
	MsgBody string
}

type tTPCChans struct {
	chSend chan *TChanSend
	chQuit chan bool
}

var onceTcpSvr sync.Once

func StartTcpSvr() {
	onceTcpSvr.Do(func() {
		port := ":" + configsvr.Fetch("tcpsvr_port")
		if ":" == strings.TrimSpace(port) {
			port = ":9988" //default 9988
		}
		l4g.Debug("tcpsvr listen on port:%s", port)
		l, err := net.Listen("tcp", port)
		if err != nil {
			l4g.Error("Failure to listen: %s", err.Error())
			return
		}
		l4g.Debug("tcpsvr::listening......")

		for {
			l4g.Debug("Started!")
			if c, err := l.Accept(); err == nil {
				l4g.Debug("Accept()::", c)
				go StartChild(c) //new thread
			}
		}
		l4g.Debug("Everything is done!")

	})
}

func StartChild(c net.Conn) {
	l4g.Debug("tcpsvr::StartChild()::Started!")
	var clientRData *proto.TDatagram = nil
	defer func() {
		if nil != clientRData {
			Logout(clientRData.GetClientId())
		} else {
			l4g.Debug("clientRData is nil!")
		}
		c.Close()
		l4g.Debug("tcpsvr::StartChild()::Ended!")
	}()

	controller := &tTPCChans{
		chSend: make(chan *TChanSend),
		chQuit: make(chan bool),
	}

	go func() { // Read
		//l4g.Debug("tcpsvr::StartChild()::Read()::begin!()")
		reader := bufio.NewReaderSize(c, 256)
		defer func() {
			close(controller.chSend)
			close(controller.chQuit)
			//l4g.Debug("tcpsvr::StartChild()::Read()::ended!()")
		}()
		for {
			// Read datagram length
			//l4g.Debug("tcpsvr::StartChild()::Read()::tried to Peek(8)")
			hlen, err := reader.Peek(8)
			if err != nil {
				l4g.Error("Failure to read datagram length:", err)
				break
			}
			l4g.Debug("tcpsvr::StartChild()::Read():datagram length:\"%s\"", string(hlen))
			// Read all datagram
			datalen, err := strconv.ParseUint(strings.TrimSpace(string(hlen)), 10, 64)
			if err != nil {
				l4g.Error("Failure to parse datagram length:", err)
				break
			}
			rowdata := make([]byte, 8+datalen)
			//l4g.Debug("tcpsvr::StartChild()::Read()::tried to read(%d)", 8+datalen)
			_, err = reader.Read(rowdata)
			if err != nil {
				l4g.Error("Fatal error", err)
				break
			}
			//l4g.Debug("Read:Data:%d:\"%s\"", n, string(rowdata))

			// Set timeout
			deadline := time.Now().Add(TCP_READ_TIMEOUT)
			c.SetReadDeadline(deadline)

			if nil == clientRData {
				//l4g.Debug("tcpsvr::StartChild()::Read()::nil == clientRData")
				if proto.MAX_BODYSIZE >= datalen && proto.MIN_BODYSIZE <= datalen {
					clientRData = proto.Decode(rowdata, datalen)
					if nil != clientRData {
						//l4g.Debug(clientRData)
						if clientRData.IsRegisterDatagram() {
							//l4g.Debug("IsRegisterDatagram!")
							// Register
							Login(clientRData.GetClientId(), controller)
							// Send Ack
							regAck := proto.MakeRegAckDatagram(clientRData.GetClientToken(), clientRData.GetClientId())
							c.Write([]byte(regAck))
							// Send messages
							Msgs := msgstore.GetUnsent(clientRData.GetClientId())
							for i := 0; i < len(Msgs); i++ {
								//msgD := proto.MakeMsgDatagram(clientRData, Msgs[i].Id, Msgs[i].Body)
								//if "" != msgD {
								//l4g.Debug(Msgs[i].Id, Msgs[i].Body)
								controller.Send(Msgs[i].Id, Msgs[i].Body)
								//}
							}
							continue
						} else {
							clientRData = nil
						}
					}
				}
			} else {
				//l4g.Debug("tcpsvr::StartChild()::Read()::nil != clientRData")
				if proto.MAX_BODYSIZE >= datalen && proto.MIN_BODYSIZE <= datalen {
					ack := proto.Decode(rowdata, datalen)
					if nil != ack {
						//l4g.Debug(ack)
						if ack.IsMsgAckDatagram() {
							//l4g.Debug("IsMsgAckDatagram!(clientid:%s,MsgId:%s)", clientRData.GetClientId(), ack.GetMessageId())
							// Send message ack
							msgstore.SendAck(clientRData.GetClientId(), ack.GetMessageId())
						}
					}
				}
			}
		}
		controller.Quit()
	}()

	// Write
	//l4g.Debug("tcpsvr::StartChild()::Write()::begin!()")
	writer := bufio.NewWriterSize(c, 256)
WriteLoop:
	for {
		select {
		case value := <-controller.chSend:
			msgD := proto.MakeMsgDatagram(string(clientRData.ClientID[:20]), value.MsgId, value.MsgBody)
			if "" != msgD {
				//l4g.Debug("Write:msg:", msgD)
				_, err := writer.Write([]byte(msgD))
				if err != nil {
					l4g.Error("Failure to write:", err)
					break WriteLoop
				}
				writer.Flush()
				//l4g.Debug("Write:", value)
			}
		case <-controller.chQuit:
			//l4g.Debug("StartChild()::received a quit message!")
			break WriteLoop
		}
	}
	//l4g.Debug("tcpsvr::StartChild()::Write()::end!()")
}

func (o *tTPCChans) Send(msgId, message string) {
	o.chSend <- &TChanSend{
		MsgId:   msgId,
		MsgBody: message,
	}
}

func (o *tTPCChans) Quit() {
	o.chQuit <- true
}

一直读取消息，正常情况下收到两种消息

1. 客户端注册
2. 客户端收到推送消息，包含clientID，messageID

读取登录数据，调用控制中心接口，把自己先注册到本地，再保存到 socket storage server。key ： clientID，value：发消息channel

goroutine休眠，等待发送消息到手机端

5.socket字典共享（redis）存储 socket storage server

pushservice\sockstore\sockstore.go

type THostname struct {
	Hostname string
	IsLocal  bool
}

type tChanGet struct {
	Key          string
	ReplyChannel chan *THostname
}

type tChanSet struct {
	clientId string
	time     int64
}

type TKeyValue struct {
	Key   string
	Value string
}

var once sync.Once

type tSockStoreChans struct {
	chGet         chan *tChanGet
	chLogin       chan *tChanSet
	chLogout      chan string
	chListAll     chan chan []string
	chListAllData chan chan []*TKeyValue
}

var sockStoreChans = &tSockStoreChans{
	chGet:         make(chan *tChanGet, 10),
	chLogin:       make(chan *tChanSet, 10),
	chLogout:      make(chan string),
	chListAll:     make(chan chan []string),
	chListAllData: make(chan chan []*TKeyValue),
}

func Start() {
	once.Do(func() {
		//index := int32(1)
		l4g.Debug("sockstore::Start()")
		var hostname string
		var client redis.Client
		//hostname, _ = os.Hostname()
		hostname = fmt.Sprintf("%s:%s", configsvr.Fetch("hostname"), configsvr.Fetch("rpcsvr_port"))
		l4g.Debug("hostname: %s", hostname)
		l4g.Debug("addr_redis: %s", configsvr.Fetch("addr_redis"))

		client.Addr = configsvr.Fetch("addr_redis")
		client.Db = 0

		defer func() {
			close(sockStoreChans.chGet)
			close(sockStoreChans.chLogin)
			close(sockStoreChans.chLogout)
			close(sockStoreChans.chListAll)
			close(sockStoreChans.chListAllData)
		}()

		for {
			select {
			case value := <-sockStoreChans.chGet:
				val, _ := client.Get(value.Key)
				l4g.Debug("%s: %s", value.Key, string(val))
				value.ReplyChannel <- &THostname{
					Hostname: string(val),
					IsLocal:  (string(val) == hostname),
				}
			case replyChannel := <-sockStoreChans.chListAll:
				l4g.Debug("sockStoreChans.chListAll")
				keys, _ := client.Keys("*")
				if len(keys) > 500 {
					replyChannel <- keys[:500]
				} else {
					replyChannel <- keys
				}
			case replyChannel := <-sockStoreChans.chListAllData:
				l4g.Debug("sockStoreChans.chListAllData")
				keys, _ := client.Keys("*")
				result := []*TKeyValue{}
				for i, el := range keys {
					if i > 499 {
						break
					}
					val, _ := client.Get(el)
					result = append(result, &TKeyValue{el, string(val)})
				}
				replyChannel <- result
			case value := <-sockStoreChans.chLogin:
				l4g.Debug("sockStoreChans.chLogin %v", value.clientId)
				if err := client.Set(value.clientId, []byte(hostname)); err != nil {
					l4g.Error(err)
				}
			case value := <-sockStoreChans.chLogout:
				l4g.Debug("sockStoreChans.chLogout %v", value)
				_, err := client.Del(value)
				if err != nil {
					l4g.Error(err)
				}
			}
		}
	})
}

// 获得clientid在线的socket的所在服务器, 第二个参数返回是否是本机
func GetHostname(key string) (string, bool) {
	sendData := &tChanGet{
		Key:          key,
		ReplyChannel: make(chan *THostname),
	}
	defer func() {
		close(sendData.ReplyChannel)
	}()
	sockStoreChans.chGet <- sendData
	result := <-sendData.ReplyChannel
	return result.Hostname, result.IsLocal
}

func Login(clientId string) {
	l4g.Debug("Login ClientID:\"%s\"", clientId)
	sockStoreChans.chLogin <- &tChanSet{
		clientId: clientId,
		time:     0,
	}
}

func Logout(clientId string) {
	l4g.Debug("Logout ClientID:\"%s\"", clientId)
	sockStoreChans.chLogout <- clientId
}

func ListAll() []string {
	ReplyChannel := make(chan []string)
	defer func() {
		close(ReplyChannel)
	}()
	sockStoreChans.chListAll <- ReplyChannel
	return <-ReplyChannel
}

func ListAllData() []*TKeyValue {
	ReplyChannel := make(chan []*TKeyValue)
	defer func() {
		close(ReplyChannel)
	}()
	sockStoreChans.chListAllData <- ReplyChannel
	return <-ReplyChannel
}

提供查询clientID在哪个服务器上，同时告诉查询方，是不是在本机

6.消息存储（redis） message storage server

pushservice\msgstore\msgstore.go

package msgstore

type TKeyValue struct {
	Key   string
	Value string
}

type TMessage struct {
	Id   string
	Body string
}

type tChanGetUnsent struct {
	ClientId  string
	ReplyChan chan []*TMessage
}

type tChanSendAck struct {
	ClientId  string
	MessageId string
}

type tChanSet struct {
	clientId string
	msgId    string
	msgBody  string
	time     int64
}

var once sync.Once

type tSockStoreChans struct {
	chGetUnsent   chan *tChanGetUnsent
	chAddMessage  chan *tChanSet
	chListAllData chan chan []*TKeyValue
	chSendAck     chan *tChanSendAck
}

var msgStoreChans = &tSockStoreChans{
	chGetUnsent:   make(chan *tChanGetUnsent, 10),
	chAddMessage:  make(chan *tChanSet, 10),
	chListAllData: make(chan chan []*TKeyValue, 10),
	chSendAck:     make(chan *tChanSendAck, 10),
}

func Start() {
	once.Do(func() {
		//index := int32(1)
		l4g.Debug("msgstore::Start()")
		var client redis.Client
		l4g.Debug("addr_redis: %s", configsvr.Fetch("addr_redis"))
		client.Addr = configsvr.Fetch("addr_redis")
		client.Db = 1
		defer func() {
			close(msgStoreChans.chGetUnsent)
			close(msgStoreChans.chAddMessage)
			close(msgStoreChans.chListAllData)
			close(msgStoreChans.chSendAck)
		}()
		for {
			select {
			case value := <-msgStoreChans.chGetUnsent:
				key := "C_" + value.ClientId
				val, err := client.Lrange(key, 0, -1)
				if err != nil {
					l4g.Error(err)
				} else {
					l4g.Debug("lrange value: %v", val)
				}
				result := []*TMessage{}
				for _, el := range val {
					mVal, err := client.Get("M_" + string(el))
					if err != nil {
						l4g.Error(err)
						client.Lrem1(key, 0, el)
					} else {
						result = append(result, &TMessage{
							Id:   string(el),
							Body: string(mVal),
						})
					}
				}
				value.ReplyChan <- result
			case value := <-msgStoreChans.chAddMessage:
				l4g.Debug("", value.clientId)
				client.Setex("M_"+value.msgId, value.time, []byte(value.msgBody))
				client.Rpush("C_"+value.clientId, []byte(value.msgId))
			case replyChannel := <-msgStoreChans.chListAllData:
				l4g.Debug("msgStoreChans.chListAllData")
				keys, _ := client.Keys("M_*")
				result := []*TKeyValue{}
				for i, el := range keys {
					if i > 299 {
						break
					}
					val, _ := client.Get(el)
					result = append(result, &TKeyValue{el, string(val)})
				}

				keys2, _ := client.Keys("C_*")
				for i, el := range keys2 {
					if i > 299 {
						break
					}
					val, _ := client.Lrange(el, 0, -1)
					strs := []string{}
					for _, s := range val {
						strs = append(strs, string(s))
					}
					result = append(result, &TKeyValue{el, strings.Join(strs, ", ")})
				}
				replyChannel <- result
			case value := <-msgStoreChans.chSendAck:
				client.Del("M_" + value.MessageId)
				client.Lrem1("C_"+value.ClientId, 0, []byte(value.MessageId))

			}
		}
	})
}

// 推送消息，保存到redis
func AddMessage(clientId, msgId, msgBody string, time int64) {
	l4g.Debug("msgstore.AddMessage()::ClientID:[%s],msgId:[%s],msgBody[%s]",
		clientId,
		msgId,
		msgBody)
	msgStoreChans.chAddMessage <- &tChanSet{
		clientId: clientId,
		msgId:    msgId,
		msgBody:  msgBody,
		time:     time,
	}
}

func ListAllData() []*TKeyValue {
	ReplyChann := make(chan []*TKeyValue)
	defer func() {
		close(ReplyChann)
	}()
	msgStoreChans.chListAllData <- ReplyChann
	return <-ReplyChann
}

// 在登录之后，获取未发送消息。从 C_clientId 活动 message id list，获得所有message，
// 如果message id已经过期，从C_clientId 中删除。返回所有message
// TODO: 返回有限条记录，删除过多的message id
func GetUnsent(clientId string) []*TMessage {
	l4g.Debug("msgstore.GetUnsent()::ClientID:[%s]", clientId)
	sendData := &tChanGetUnsent{
		ClientId:  clientId,
		ReplyChan: make(chan []*TMessage),
	}
	defer func() {
		close(sendData.ReplyChan)
	}()
	msgStoreChans.chGetUnsent <- sendData
	return <-sendData.ReplyChan
}

// 收到消息送达的通知，删除redis的对应记录
func SendAck(clientId, msgId string) {
	l4g.Debug("msgstore.SendAck()::ClientID:[%s],msgId:[%s]", clientId, msgId)
	msgStoreChans.chSendAck <- &tChanSendAck{
		ClientId:  clientId,
		MessageId: msgId,
	}
}

clientID, messageID, messageBody 三级存储

提供未发送消息的查询

收到消息发送成功的回调，删除消息

g

g

g

g

g

gg

7.连接兄弟节点的rpc client

pushservice\channelsvr\rpcclient.go

var remoteCallOnce sync.Once

type TRemotePushMessage struct {
	RemoteAddress string
	TPushMsgArgs
}

type TRemoteCallChans struct {
	chPushMessage chan *TRemotePushMessage
}

var remoteCallChans = &TRemoteCallChans{
	chPushMessage: make(chan *TRemotePushMessage, 10),
}

func RpcClientStart() {
	remoteCallOnce.Do(func() {
		l4g.Debug("channelsvr.RpcClientStart()")
		data := map[string]*rpc.Client{}
		defer func() {
			close(remoteCallChans.chPushMessage)
		}()

		FReconn := func(remoteaddr string) *rpc.Client {
			client, err := rpc.DialHTTP("tcp", remoteaddr)
			if err == nil {
				data[remoteaddr] = client
			} else {
				l4g.Error("dialing:", err)
			}
			return client
		}

		for {
			select {
			case value := <-remoteCallChans.chPushMessage:
				l4g.Debug("remoteCallChans.chPushMessage: %s", value.RemoteAddress)
				client, ok := data[value.RemoteAddress]
				if !ok {
					client = FReconn(value.RemoteAddress)
				}
				if client != nil {
					FCall := func(method string, args *TPushMsgArgs, reply *string) error {
						err := client.Call(method, args, &reply)
						if err != nil {
							l4g.Error("%s error: %s", method, err)
						}
						return err
					}
					var reply string
					err := FCall("Channel.PushMsg", &value.TPushMsgArgs, &reply)
					if err != nil {
						//有可能因为超时连接断掉，重试一次
						client = FReconn(value.RemoteAddress)
						FCall("Channel.PushMsg", &value.TPushMsgArgs, &reply)
					}
				}
			}
		}
	})
}

func RemotePushMessage(remoteAddr, clientId, msgId, msg string) {
	sendData := &TRemotePushMessage{
		RemoteAddress: remoteAddr,
		TPushMsgArgs: TPushMsgArgs{
			ClientID:  clientId,
			MessageID: msgId,
			Message:   msg,
		},
	}
	remoteCallChans.chPushMessage <- sendData
}

共享连接，考虑断线重连机制

重要的api就是推送消息

8.rpc server

pushservice\channelsvr\rpc_channelsvr.go

var rpcServerOnce sync.Once

type Config int

func (t *Config) Fetch(arg string, reply *string) error {
	l4g.Debug("call rpcsvr.Fetch %s", arg)
	*reply = configsvr.Fetch(arg)
	return nil
}

type Channel int
type TPushMsgArgs struct {
	ClientID  string
	MessageID string
	Message   string
}

func (t *Channel) PushMsg(args TPushMsgArgs, reply *string) error {
	l4g.Debug("call Channel.PushMsg %s, %s", args.ClientID, args.Message)
	PushLocalMessage(args.ClientID, args.MessageID, args.Message)
	*reply = "Ok"
	return nil
}

func (t *Channel) Broadcast(args string, reply *string) error {
	l4g.Debug("call Channel.Broadcast %s", args)
	Broadcast(args)
	*reply = "Ok"
	return nil
}

func (t *Channel) FindClient(clientID string, reply *string) error {
	l4g.Debug("call Channel.GetChannel %s", clientID)
	if nil == GetChannel(clientID) {
		*reply = "Unfound"
	} else {
		*reply = "Ok"
	}
	return nil
}

// 获得本机在线人数
func (t *Channel) GetCount(args string, reply *int) error {
	l4g.Debug("call Channel.GetCount")
	*reply = GetCount()
	return nil
}

func RpcServerStart() {
	rpcServerOnce.Do(func() {
		l4g.Debug("rpcsvr::Start()")
		config := new(Config)
		rpc.Register(config)
		channel := new(Channel)
		rpc.Register(channel)
		sockStore := new(SockStore)
		rpc.Register(sockStore)
		messageStore := new(MessageStore)
		rpc.Register(messageStore)
		rpc.HandleHTTP()
		port := ":" + configsvr.Fetch("rpcsvr_port")
		if ":" == strings.TrimSpace(port) {
			port = ":1234" //default 1234
		}
		l4g.Debug("rpcsvr listen on port:%s", port)
		l, e := net.Listen("tcp", port)
		if e != nil {
			l4g.Error("listen error:", e)
		}
		http.Serve(l, nil)
	})
}

远程过来的请求，推送本地消息

9.对外消息接口

pushservice\channelsvr\rpc_msgstore.go

type MessageStore int

func (t *MessageStore) ListAllData(arg string, reply *[]*msgstore.TKeyValue) error {
	l4g.Debug("call MessageStore.ListAllData %s", arg)
	*reply = msgstore.ListAllData()
	return nil
}

type TAddMessageArg struct {
	ClientId    string
	MessageId   string
	MessageBody string
	Time        int64
}

func (t *MessageStore) AddMessage(arg *TAddMessageArg, reply *string) error {
	l4g.Debug("call MessageStore.AddMessage %v", arg)
	msgstore.AddMessage(arg.ClientId, arg.MessageId, arg.MessageBody, arg.Time)
	PushMessage(arg.ClientId, arg.MessageId, arg.MessageBody)
	*reply = arg.MessageId
	return nil
}

func (t *MessageStore) GetUnsent(arg string, reply *[]*msgstore.TMessage) error {
	l4g.Debug("call MessageStore.GetUnsent %s", arg)
	*reply = msgstore.GetUnsent(arg)
	return nil
}

type TSendAckArg struct {
	ClientId  string
	MessageId string
}

func (t *MessageStore) SendAck(arg *TSendAckArg, reply *string) error {
	l4g.Debug("call MessageStore.SendAck %s", arg)
	msgstore.SendAck(arg.ClientId, arg.MessageId)
	*reply = arg.MessageId
	return nil
}

重要api，AddMessage

存储消息，然后发送消息，发送不成功，消息还在，等待上线推送