notify/server.go

// Package notify is a package which provide common tcp/udp/unix socket service
package notify

import (
	"net"
	"time"

	"b612.me/starainrt"
)

var builder *starainrt.StarQueue

func init() {
	builder = starainrt.NewQueue()
}

// StarNotifyS 为Server端
type StarNotifyS struct {
	// Queue 是用来处理收发信息的简单消息队列
	Queue *starainrt.StarQueue
	// FuncLists 记录了被通知项所记录的函数
	FuncLists      map[string]func(NetMsg) string
	defaultFunc    func(NetMsg) string
	serverStopSign chan int
	notifychan     chan int
}

// NetMsg 指明当前被通知的关键字
type NetMsg struct {
	Conn net.Conn
	key  string
}

// Send 用于向client端发送数据
func (nmsg *NetMsg) Send(msg string) error {
	_, err := nmsg.Conn.Write(builder.BuildMessage(nmsg.key + "||" + msg))
	return err
}

func (star *StarNotifyS) starinits() {
	star.serverStopSign, star.notifychan = make(chan int), make(chan int, 3)
	star.Queue = starainrt.NewQueue()
	star.FuncLists = make(map[string]func(NetMsg) string)
}

// NewNotifyS 开启一个新的Server端通知
func NewNotifyS(netype, value string) (StarNotifyS, error) {
	var star StarNotifyS
	star.starinits()
	listener, err := net.Listen(netype, value)
	if err == nil {
		go star.notify()
		go func() {
			for {
				select {
				case <-star.serverStopSign:
					star.notifychan <- 1
					listener.Close()
					return
				default:
				}
				conn, err := listener.Accept()
				if err != nil {
					continue
				}
				go func(conn net.Conn) {
					for {
						select {
						case <-star.serverStopSign:
							star.notifychan <- 1
							conn.Close()
							return
						default:
						}
						buf := make([]byte, 8192)
						n, err := conn.Read(buf)
						if n != 0 {
							star.Queue.ParseMessage(buf[0:n], conn)
						}
						if err != nil {
							conn.Close()
							break
						}
					}
				}(conn)
			}
		}()
	}
	return star, err
}

// SetNotify 用于设置通知关键词的调用函数
func (star *StarNotifyS) SetNotify(name string, data func(NetMsg) string) {
	star.FuncLists[name] = data
}

// SetDefaultNotify 用于设置默认关键词的调用函数
func (star *StarNotifyS) SetDefaultNotify(name string, data func(NetMsg) string) {
	star.defaultFunc = data
}

func (star *StarNotifyS) notify() {
	for {
		select {
		case <-star.notifychan:
			return
		default:
		}
		data, err := star.Queue.RestoreOne()
		if err != nil {
			time.Sleep(time.Millisecond * 20)
			continue
		}
		if msg, ok := star.FuncLists[data.Msg]; ok {
			sdata := msg(NetMsg{data.Conn.(net.Conn), data.Msg})
			if sdata == "" {
				continue
			}
			sdata = data.Msg + "||" + sdata
			data.Conn.(net.Conn).Write(star.Queue.BuildMessage(sdata))
		} else {
			if star.defaultFunc != nil {
				sdata := star.defaultFunc(NetMsg{data.Conn.(net.Conn), data.Msg})
				if sdata == "" {
					continue
				}
				sdata = data.Msg + "||" + sdata
				data.Conn.(net.Conn).Write(star.Queue.BuildMessage(sdata))
			}
		}
	}
}

// ServerStop 用于终止Server端运行
func (star *StarNotifyS) ServerStop() {
	star.serverStopSign <- 0
}