cometclient.go

package main

import (
	"bufio"
	log "code.google.com/p/log4go"
	"fmt"
	"net"
	"strconv"
	"time"
)

func Cometclient(aaddr, key string, heartbeat int) {
	addr, err := net.ResolveTCPAddr("tcp", aaddr)
	if err != nil {
		log.Error("net.ResolveTCPAddr(\"tcp\", \"%s\") failed (%s)", aaddr, err.Error())
		return
	}
	log.Info("connect to gopush-cluster comet")
	conn, err := net.DialTCP("tcp", nil, addr)
	if err != nil {
		log.Error("net.DialTCP() failed (%s)", err.Error())
		return
	}
	log.Info("send sub request")

	proto := []byte(fmt.Sprintf("*3\r\n$3\r\nsub\r\n$%d\r\n%s\r\n$%d\r\n%d\r\n", len(key), key, len(strconv.Itoa(int(heartbeat))), heartbeat))
	log.Info("send protocol: %s", string(proto))
	if _, err := conn.Write(proto); err != nil {
		log.Error("conn.Write() failed (%s)", err.Error())
		return
	}
	// get first heartbeat
	first := false
	rd := bufio.NewReader(conn)
	// block read reply from service
	log.Info("wait message")
	for {
		if err := conn.SetReadDeadline(time.Now().Add(time.Second * time.Duration(heartbeat) * 2)); err != nil {
			log.Error("conn.SetReadDeadline() failed (%s)", err.Error())
			return
		}
		line, err := rd.ReadBytes('\n')
		if err != nil {
			log.Error("rd.ReadBytes() failed (%s)", err.Error())
			return
		}
		if line[len(line)-2] != '\r' {
			log.Error("protocol reply format error")
			return
		}
		log.Info("line: %s", line)
		switch line[0] {
		// reply
		case '$':
			cmdSize, err := strconv.Atoi(string(line[1 : len(line)-2]))
			if err != nil {
				log.Error("protocol reply format error")
				return
			}
			data, err := rd.ReadBytes('\n')
			if err != nil {
				log.Error("protocol reply format error")
				return
			}
			if len(data) != cmdSize+2 {
				log.Error("protocol reply format error: %s", data)
				return
			}
			if data[cmdSize] != '\r' || data[cmdSize+1] != '\n' {
				log.Error("protocol reply format error")
				return
			}
			reply := string(data[0:cmdSize])
			log.Info("receive msg: %s", reply)
			break
			// heartbeat
		case '+':
			if !first {
				// send heartbeat
				go func() {
					for {
						log.Info("send heartbeat")
						if _, err := conn.Write([]byte("h")); err != nil {
							log.Error("conn.Write() failed (%s)", err.Error())
							return
						}
						time.Sleep(time.Duration(heartbeat) * time.Second)
					}
				}()
				first = true
			}
			log.Info("receive heartbeat")
			break
		}
	}
}