outdate
兔子 5 years ago
parent 6004b3a0f4
commit dc7e6db1a0

@ -10,13 +10,13 @@ import (
// StarNotifyC 为Client端 // StarNotifyC 为Client端
type StarNotifyC struct { type StarNotifyC struct {
connc net.Conn Connc net.Conn
clientSign map[string]chan string clientSign map[string]chan string
// FuncLists 当不使用channel时使用此记录调用函数 // FuncLists 当不使用channel时使用此记录调用函数
FuncLists map[string]func(CMsg) FuncLists map[string]func(CMsg)
clientStopSign chan int clientStopSign chan int
defaultFunc func(CMsg) defaultFunc func(CMsg)
// Stop 停止信 // Stop 停止信
Stop chan int Stop chan int
// UseChannel 是否使用channel作为信息传递 // UseChannel 是否使用channel作为信息传递
UseChannel bool UseChannel bool
@ -71,34 +71,34 @@ func NewNotifyC(netype, value string) (*StarNotifyC, error) {
if strings.Index(netype, "udp") >= 0 { if strings.Index(netype, "udp") >= 0 {
star.isUDP = true star.isUDP = true
} }
star.connc, err = net.Dial(netype, value) star.Connc, err = net.Dial(netype, value)
if err != nil { if err != nil {
return nil, err return nil, err
} }
go star.cnotify() go star.cnotify()
go func() {
for {
go func() { go func() {
<-star.clientStopSign <-star.clientStopSign
star.notifychan <- 2 star.notifychan <- 2
star.connc.Close() star.Connc.Close()
star.Stop <- 0 star.Stop <- 0
star.Online = false star.Online = false
return return
}() }()
go func() {
for {
buf := make([]byte, 8192) buf := make([]byte, 8192)
n, err := star.connc.Read(buf) n, err := star.Connc.Read(buf)
if n != 0 {
star.Queue.ParseMessage(buf[0:n], star.Connc)
}
if err != nil { if err != nil {
star.connc.Close() star.Connc.Close()
star.Stop <- 1 star.Stop <- 1
star.notifychan <- 3 star.notifychan <- 3
//star, _ = NewNotifyC(netype, value) //star, _ = NewNotifyC(netype, value)
star.Online = false star.Online = false
return return
} }
if n != 0 {
star.Queue.ParseMessage(buf[0:n], star.connc)
}
} }
}() }()
star.Online = true star.Online = true
@ -107,13 +107,13 @@ func NewNotifyC(netype, value string) (*StarNotifyC, error) {
// Send 用于向Server端发送数据 // Send 用于向Server端发送数据
func (star *StarNotifyC) Send(name string) error { func (star *StarNotifyC) Send(name string) error {
_, err := star.connc.Write(star.Queue.BuildMessage(name)) _, err := star.Connc.Write(star.Queue.BuildMessage(name))
return err return err
} }
// SendValue 用于向Server端发送key-value类型数据 // SendValue 用于向Server端发送key-value类型数据
func (star *StarNotifyC) SendValue(name, value string) error { func (star *StarNotifyC) SendValue(name, value string) error {
_, err := star.connc.Write(star.Queue.BuildMessage(name + "||" + value)) _, err := star.Connc.Write(star.Queue.BuildMessage(name + "||" + value))
return err return err
} }

@ -31,6 +31,13 @@ type StarNotifyS struct {
UDPConn *net.UDPConn UDPConn *net.UDPConn
// Online 当前链接是否处于活跃状态 // Online 当前链接是否处于活跃状态
Online bool Online bool
// ReadDeadline tcp/unix中读超时设置,udp请直接调用UDPConn
ReadDeadline time.Time
// WriteDeadline tcp/unix中写超时设置,udp请直接调用UDPConn
WriteDeadline time.Time
// Deadline tcp/unix中超时设置,udp请直接调用UDPConn
Deadline time.Time
} }
// SMsg 指明当前服务端被通知的关键字 // SMsg 指明当前服务端被通知的关键字
@ -94,8 +101,6 @@ func doudps(netype, value string) (*StarNotifyS, error) {
return nil, err return nil, err
} }
go star.notify() go star.notify()
go func() {
for {
go func() { go func() {
<-star.serverStopSign <-star.serverStopSign
star.notifychan <- 1 star.notifychan <- 1
@ -108,6 +113,7 @@ func doudps(netype, value string) (*StarNotifyS, error) {
star.Online = false star.Online = false
return return
}() }()
go func() {
for { for {
buf := make([]byte, 8192) buf := make([]byte, 8192)
n, addr, err := star.UDPConn.ReadFromUDP(buf) n, addr, err := star.UDPConn.ReadFromUDP(buf)
@ -119,7 +125,6 @@ func doudps(netype, value string) (*StarNotifyS, error) {
continue continue
} }
} }
}
}() }()
star.Online = true star.Online = true
return &star, nil return &star, nil
@ -134,8 +139,6 @@ func notudps(netype, value string) (*StarNotifyS, error) {
return nil, err return nil, err
} }
go star.notify() go star.notify()
go func() {
for {
go func() { go func() {
<-star.serverStopSign <-star.serverStopSign
star.notifychan <- 3 star.notifychan <- 3
@ -148,7 +151,8 @@ func notudps(netype, value string) (*StarNotifyS, error) {
star.Online = false star.Online = false
return return
}() }()
go func() {
for {
conn, err := listener.Accept() conn, err := listener.Accept()
if err != nil { if err != nil {
select { select {
@ -159,6 +163,15 @@ func notudps(netype, value string) (*StarNotifyS, error) {
continue continue
} }
} }
if !star.ReadDeadline.IsZero() {
conn.SetReadDeadline(star.ReadDeadline)
}
if !star.WriteDeadline.IsZero() {
conn.SetWriteDeadline(star.WriteDeadline)
}
if !star.Deadline.IsZero() {
conn.SetDeadline(star.Deadline)
}
go func(conn net.Conn) { go func(conn net.Conn) {
for { for {
buf := make([]byte, 8192) buf := make([]byte, 8192)
@ -213,21 +226,23 @@ func (star *StarNotifyS) notify() {
continue continue
} }
} }
go func() {
if msg, ok := star.FuncLists[key]; ok { if msg, ok := star.FuncLists[key]; ok {
sdata := msg(rmsg) sdata := msg(rmsg)
if sdata == "" { if sdata == "" {
continue return
} }
rmsg.Reply(sdata) rmsg.Reply(sdata)
} else { } else {
if star.defaultFunc != nil { if star.defaultFunc != nil {
sdata := star.defaultFunc(rmsg) sdata := star.defaultFunc(rmsg)
if sdata == "" { if sdata == "" {
continue return
} }
rmsg.Reply(sdata) rmsg.Reply(sdata)
} }
} }
}()
} }
} }

Loading…
Cancel
Save