.

git-svn-id: svn://47.119.165.148/zhub@63 e63fbceb-bcc3-4977-ac22-735b83d8d0f4

cli/zdb-client.go

@@ -14,9 +14,10 @@ import (
 
 var (
 	reconnect    = 0
-	subEvent     = make(map[string]func(v string))
-	chReceive    = make(chan []string, 1000)
+	subFun       = make(map[string]func(v string))
+	timerFun     = make(map[string]func())
 	chSend       = make(chan []string, 1000)
+	chReceive    = make(chan []string, 1000)
 	timerReceive = make(chan []string, 1000)
 )
 
@@ -57,7 +58,7 @@ func (c *Client) reconn() (err error) {
 		} else if err == nil {
 			c.conn = conn
 			go c.receive()
-			for topic, _ := range subEvent {
+			for topic, _ := range subFun {
 				c.subscribes(topic)
 			}
 			break
@@ -72,38 +73,29 @@ func (c *Client) init() {
 		for {
 			select {
 			case vs := <-chReceive:
-				fun := subEvent[vs[1]]
+				fun := subFun[vs[1]]
 				if fun == nil {
 					log.Println("topic received, nothing to do", vs[1], vs[2])
 					continue
 				}
 				fun(vs[2])
 			case vs := <-timerReceive:
-				log.Println("收到 timer 消息 ", vs[1])
+				fun := timerFun[vs[1]]
+				if fun == nil {
+					log.Println("timer received, nothing to do", vs[1])
+					continue
+				}
+				fun()
 			}
 
 		}
-
-		/*for {
-			vs, ok := <-chReceive
-			if !ok {
-				break
-			}
-
-			fun := subEvent[vs[1]]
-			if fun == nil {
-				log.Println("topic received, nothing to do", vs[1], vs[2])
-				continue
-			}
-			fun(vs[2])
-		}*/
 	}()
 
 	go c.receive()
 }
 
 func (c *Client) Subscribe(topic string, fun func(v string)) {
-	subEvent[topic] = fun
+	subFun[topic] = fun
 	c.subscribes(topic)
 }
 
@@ -139,7 +131,8 @@ func (c *Client) Daly(topic string, message string, daly int) error {
 	return nil
 }
 
-func (c *Client) Timer(topic string, expr string) {
+func (c *Client) Timer(topic string, expr string, fun func()) {
+	timerFun[topic] = fun
 	c.send("timer", topic, expr)
 }
 

cli/zdb-client_test.go

@@ -2,11 +2,8 @@ package cli
 
 import (
 	"testing"
-
 )
 
 func TestClient(t *testing.T) {
 
-
-
 }

cli_test.go

@@ -8,22 +8,25 @@ import (
 	"zhub/cli"
 )
 
-func TestName(t *testing.T) {
+func TestCli(t *testing.T) {
 	//client, err := cli.Create("39.108.56.246:1216", "")
 	client, err := cli.Create("127.0.0.1:1216", "")
+
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	client.Subscribe("a-1", func(v string) {
-		log.Println(v)
+	// 订阅主题 消息
+	client.Subscribe("a", func(v string) {
+		log.Println("收到主题 a 消息 " + v)
 	})
 
+	//
 	client.Timer("t", "* * * * * *")
 
 	go func() {
 		for i := 0; i < 50000; i++ {
-			client.Publish("a-1", strconv.Itoa(i))
+			client.Publish("a", strconv.Itoa(i))
 			time.Sleep(time.Second)
 		}
 	}()

main.go

@@ -6,7 +6,7 @@ import (
 	"strconv"
 	"strings"
 	"zhub/cli"
-	"zhub/zdb"
+	"zhub/zsub"
 )
 
 func main() {
@@ -30,7 +30,7 @@ func main() {
 	}
 
 	if server {
-		zdb.ServerStart(host, port)
+		zsub.ServerStart(host, port)
 	} else {
 		cli.ClientRun(host, port)
 	}

zdb/rcmd-exec.go

@@ -1,4 +1,4 @@
-package zdb
+package _zdb
 
 import (
 	"log"
@@ -9,10 +9,10 @@ import (
 	"time"
 )
 
-func execCmd(rcmd []string, conn net.Conn) {
+func ExecCmd(rcmd []string, conn net.Conn) {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Println("execCmd Recovered:", r)
+			log.Println("ExecCmd Recovered:", r)
 		}
 	}()
 	if len(rcmd) == 0 {
@@ -58,7 +58,7 @@ func execCmd(rcmd []string, conn net.Conn) {
 	case "daly":
 		daly(rcmd, conn)
 	case "timer":
-		Timer(rcmd, conn)
+		timer(rcmd, conn)
 	default:
 		conn.Write([]byte("-Error: default not supported:[" + strings.Join(rcmd, " ") + "]\r\n"))
 		return

zdb/zdb-server.go

@@ -1,4 +1,4 @@
-package zdb
+package _zdb
 
 import (
 	"bufio"
@@ -16,7 +16,7 @@ var (
 	zsub    = make(map[string][]*ConnContext) // topic -- connx[]
 	retOk   = []byte("+OK")
 	retHelp = []byte(
-		"\n--- zdb help ---\n" +
+		"\n--- _zdb help ---\n" +
 			"______  _____   _____  \n|___  / |  _  \\ |  _  \\ \n   / /  | | | | | |_| | \n  / /   | | | | |  _  { \n / /__  | |_| | | |_| | \n/_____| |_____/ |_____/ \n" +
 			"had supported command:\n" +
 			"1. set:\n" +
@@ -33,7 +33,7 @@ var (
 			" eg: incr a\n" +
 			"7. decr:\n" +
 			" eg: decr a\n" +
-			"--- zdb help ---\n")
+			"--- _zdb help ---\n")
 )
 
 // 数据封装
@@ -56,7 +56,7 @@ func ServerStart(host string, port int) {
 		log.Fatal(err)
 		return
 	}
-	log.Printf("zdb started listen on: %s:%d \n", host, port)
+	log.Printf("_zdb started listen on: %s:%d \n", host, port)
 
 	// 启动消息监听处理
 	go func() {
@@ -66,7 +66,7 @@ func ServerStart(host string, port int) {
 				break
 			}
 
-			execCmd(v.Rcmd, *&*v.Conn)
+			ExecCmd(v.Rcmd, *&*v.Conn)
 		}
 	}()
 
@@ -108,7 +108,6 @@ func connHandler(conn net.Conn) {
 	for {
 		rcmd := make([]string, 0)
 		line, _, err := reader.ReadLine()
-		// fmt.Println("line:", string(line)) todo 可使用第一行用于协议头
 		if err != nil {
 			log.Println(err)
 			return

zdb/zdb_test.go

@@ -1,7 +1,7 @@
-package zdb
+package _zdb
 
 import "testing"
 
-func TestName(t *testing.T) {
+func TestService(t *testing.T) {
 	ServerStart("127.0.0.1", 1216)
 }

zdb/ztimer.go

@@ -0,0 +1,62 @@
+package _zdb
+
+import (
+	"fmt"
+	"github.com/robfig/cron"
+	"net"
+	"strings"
+	"time"
+)
+
+var (
+	zTimer = make(map[string]*ZTimer)
+)
+
+type ZTimer struct {
+	conns []*net.Conn
+	expr  string
+	topic string
+	cron  *cron.Cron
+}
+
+func timer(rcmd []string, conn net.Conn) {
+	ztimer := zTimer[rcmd[1]]
+	if ztimer == nil {
+		ztimer = &ZTimer{
+			conns: []*net.Conn{},
+			topic: rcmd[1],
+		}
+		zTimer[rcmd[1]] = ztimer
+	}
+
+	_conns := make([]*net.Conn, 0)
+	for _, c := range ztimer.conns {
+		if *&conn == *c {
+			continue
+		}
+		_conns = append(_conns, c)
+	}
+	_conns = append(_conns, &conn)
+	ztimer.conns = _conns
+
+	if !strings.EqualFold(ztimer.expr, rcmd[2]) {
+		ztimer.expr = rcmd[2]
+		if ztimer.cron != nil {
+			ztimer.cron.Stop()
+		}
+		ztimer.cron = func() *cron.Cron {
+			c := cron.New()
+			c.AddFunc(ztimer.expr, func() {
+				fmt.Println(time.Now().Second())
+				for _, conn := range ztimer.conns {
+					Send(*conn, "timer", ztimer.topic)
+				}
+			})
+			go c.Run()
+			return c
+		}()
+	}
+
+	zTimer[ztimer.topic] = ztimer
+	fmt.Println("xx")
+}

zsub/msg-consumer.go

@@ -0,0 +1,108 @@
+package zsub
+
+import (
+	"log"
+	"net"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+func msgAccept(v Message) {
+	defer func() {
+		if r := recover(); r != nil {
+			log.Println("ExecCmd Recovered:", r)
+		}
+	}()
+	c := v.Conn
+	rcmd := v.Rcmd
+
+	if len(rcmd) == 0 {
+		return
+	}
+
+	log.Println("rcmd: " + strings.Join(rcmd, " "))
+
+	if len(rcmd) == 1 {
+		switch strings.ToLower(rcmd[0]) {
+		default:
+			// subscribe|unsubscribe|daly
+			if strings.Index(rcmd[0], "subscribe") == 0 || strings.Index(rcmd[0], "unsubscribe") == 0 || strings.Index(rcmd[0], "daly") == 0 {
+				rcmd = strings.Split(rcmd[0], " ")
+			} else {
+				send(c.conn, "-Error: not supported! (tips: send help)")
+				return
+			}
+		}
+	}
+
+	cmd := rcmd[0]
+	switch cmd {
+	case "subscribe":
+		//subscribe x y z
+		for _, topic := range rcmd[1:] {
+			zsub.subscribe(c, topic) // todo: 批量一次订阅
+		}
+	case "unsubscribe":
+		for _, topic := range rcmd[1:] {
+			zsub.unsubscribe(c, topic)
+		}
+	case "publish":
+		if len(rcmd) != 3 {
+			send(c.conn, "-Error: publish para number!")
+		} else {
+			zsub.publish(rcmd[1], rcmd[2])
+		}
+	case "daly":
+		daly(rcmd, c)
+	case "timer":
+		// todo Timer(rcmd, conn)
+	default:
+		send(c.conn, "-Error: default not supported:["+strings.Join(rcmd, " ")+"]")
+		return
+	}
+}
+
+// daly topic valye 100
+func daly(rcmd []string, c *ZConn) {
+	if len(rcmd) != 4 {
+		send(c.conn, "-Error: subscribe para number!")
+		return
+	}
+
+	t, err := strconv.ParseInt(rcmd[3], 10, 64)
+	if err != nil {
+		send(c.conn, "-Error: "+strings.Join(rcmd, " "))
+		return
+	}
+
+	timer := time.NewTimer(time.Duration(t) * time.Millisecond)
+	select {
+	case <-timer.C:
+		zsub.publish(rcmd[1], rcmd[2])
+	}
+}
+
+var wlock = sync.Mutex{}
+
+// 发送消息
+func send(conn *net.Conn, vs ...string) error {
+	wlock.Lock()
+	defer wlock.Unlock()
+
+	var bytes []byte
+
+	if len(vs) == 1 {
+		bytes = []byte(vs[0] + "\r\n")
+	} else if len(vs) > 1 {
+		data := "*" + strconv.Itoa(len(vs)) + "\r\n"
+		for _, v := range vs {
+			data += "$" + strconv.Itoa(len(v)) + "\r\n"
+			data += v + "\r\n"
+		}
+		bytes = []byte(data)
+	}
+	_, err := (*conn).Write(bytes)
+	return err
+}

zsub/zdb.go

@@ -0,0 +1,11 @@
+package zsub
+
+var (
+	chanMessages = make(chan Message, 1000) //接收到的 所有消息数据
+)
+
+// 数据封装
+type Message struct {
+	Conn *ZConn
+	Rcmd []string
+}

zsub/zsub.go

@@ -1,7 +1,11 @@
 package zsub
 
 import (
+	"bufio"
+	"fmt"
+	"log"
 	"net"
+	"strconv"
 	"sync"
 )
 
@@ -12,13 +16,14 @@ var (
 type ZSub struct {
 	sync.Mutex
 	topics map[string]*ZTopic
+	timers map[string]*ZTimer
 }
 
-
 type ZConn struct { //ZConn
 	conn    *net.Conn
 	groupid string
 	topics  []string
+	timers  []string // 订阅、定时调度分别创建各自连接
 }
 
 /*
@@ -77,7 +82,7 @@ func (s ZSub) unsubscribe(c *ZConn, topic string) { // 取消订阅 zconn{}
 
 /*
 发送主题消息
-1、写入主题消息列表（zdb）
+1、写入主题消息列表（_zdb）
 2、回复消息写入成功
 3、推送主题消息
 */
@@ -90,6 +95,95 @@ func (s ZSub) publish(topic string, message string) {
 	}
 
 	for _, zgroup := range ztopic.groups {
-		zgroup.chMsg <- message
+		zgroup.chMsg <- message // 不同主题消费独立进行
+	}
+}
+
+func (s ZSub) close(c *ZConn) {
+	// 订阅
+	for _, topic := range c.topics {
+		s.unsubscribe(c, topic)
+	}
+
+	// 延时
+
+	// timer conn close
+	for _, topic := range c.timers { // fixme: 数据逻辑交叉循环
+		timer := s.timers[topic]
+		if timer != nil {
+			timer.close(c)
+		}
+	}
+}
+
+// ==================  ZHub 服务 =====================================
+func ServerStart(host string, port int) {
+	listen, err := net.Listen("tcp", fmt.Sprintf("%s:%d", host, port))
+	if err != nil {
+		log.Fatal(err)
+		return
+	}
+	log.Printf("_zdb started listen on: %s:%d \n", host, port)
+
+	// 启动消息监听处理
+	go func() {
+		for {
+			v, ok := <-chanMessages
+			if !ok {
+				break
+			}
+
+			// 事件消费
+			msgAccept(v)
+		}
+	}()
+
+	for {
+		conn, err := listen.Accept()
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+		fmt.Println("conn start: ", conn.RemoteAddr())
+
+		go zsub.acceptHandler(&ZConn{conn: &conn})
+	}
+}
+
+// 连接处理
+func (s ZSub) acceptHandler(c *ZConn) {
+	defer func() {
+		s.close(c) // 关闭连接
+	}()
+
+	reader := bufio.NewReader(*c.conn)
+	for {
+		rcmd := make([]string, 0)
+		line, _, err := reader.ReadLine()
+		if err != nil {
+			log.Println(err)
+			return
+		}
+		if len(line) == 0 {
+			continue
+		}
+		switch string(line[:1]) {
+		case "*":
+			n, _ := strconv.Atoi(string(line[1:]))
+			for i := 0; i < n; i++ {
+				reader.ReadLine()
+				v, _, _ := reader.ReadLine()
+				rcmd = append(rcmd, string(v))
+			}
+		default:
+			rcmd = append(rcmd, string(line))
+		}
+
+		if len(rcmd) == 0 {
+			continue
+		}
+
+		// 接收消息 zdb fixme： 细节暴露太多
+		chanMessages <- Message{Conn: c, Rcmd: rcmd}
 	}
 }

zsub/zsub_test.go

@@ -8,6 +8,7 @@ import (
 func TestName(t *testing.T) {
 	sub := ZSub{
 		topics: map[string]*ZTopic{},
+		timers: map[string]*ZTimer{},
 	}
 
 	sub.subscribe(&ZConn{

zsub/ztimer.go

@@ -0,0 +1,62 @@
+package zsub
+
+import (
+	"fmt"
+	"github.com/robfig/cron"
+	"strings"
+	"time"
+)
+
+type ZTimer struct {
+	conns []*ZConn
+	expr  string
+	topic string
+	cron  *cron.Cron
+}
+
+func (s ZSub) timer(rcmd []string, c *ZConn) {
+	timer := s.timers[rcmd[1]]
+	if timer == nil {
+		timer = &ZTimer{
+			conns: []*ZConn{},
+			topic: rcmd[1],
+		}
+		s.timers[rcmd[1]] = timer
+	}
+
+	_conns := make([]*ZConn, 0)
+	for _, conn := range timer.conns {
+		if conn == c {
+			continue
+		}
+		_conns = append(_conns, c)
+	}
+	_conns = append(_conns, c)
+	timer.conns = _conns
+
+	if !strings.EqualFold(timer.expr, rcmd[2]) {
+		timer.expr = rcmd[2]
+		if timer.cron != nil {
+			timer.cron.Stop()
+		}
+		timer.cron = func() *cron.Cron {
+			c := cron.New()
+			c.AddFunc(timer.expr, func() {
+				fmt.Println(time.Now().Second())
+				for _, conn := range timer.conns {
+					send(conn.conn, "timer", timer.topic)
+				}
+			})
+			go c.Run()
+			return c
+		}()
+	}
+
+	s.timers[rcmd[1]] = timer
+	fmt.Println("xx")
+}
+
+func (t ZTimer) close(c *ZConn) {
+	// todo timer zconn
+
+}

zsub/ztopic.go

@@ -9,6 +9,6 @@ type ZTopic struct { //ZTopic
 	chMsg  chan string // 主题消息投递
 }
 
-func createZTopic()  {
+// 主题消息发送
 
-}
+//

ztimer/ztimer.go

@@ -1,3 +0,0 @@
-package ztimer
-
-