overflow_backup_2018_0455/overflow_subscriber
4
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

ing

Browse Source
This commit is contained in:
crusader 2017-09-27 17:30:32 +09:00
parent c4b9b30ae1
commit 90b8d44773
2 changed files with 21 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
redis/subscribers.go
View File

@ -78,12 +78,12 @@ func (s *subscribers) listenSubscriptions() {
switch v := s.conn.Receive().(type) { switch v := s.conn.Receive().(type) {
case redis.Message: case redis.Message:
if h, ok := s.subHandlers[v.Channel]; ok { if h, ok := s.subHandlers[v.Channel]; ok {
var message ofs.SubscribeMessage if message, err := s.unmarshalMessage(v.Data); nil != err {
if err := json.Unmarshal(v.Data, &message); nil != err { logging.Logger.Error(fmt.Sprintf("Subscriber Unmarshal error:%v", err))
logging.Logger.Error(fmt.Sprintf("Subscriber: Unmarshal error:%v", err))
break break
} else {
h.OnSubscribe(v.Channel, message)
} }
h.OnSubscribe(v.Channel, message)
} }
break break
case redis.Subscription: case redis.Subscription:
@ -99,6 +99,20 @@ func (s *subscribers) listenSubscriptions() {
s.isListen = true s.isListen = true
} }
func (s *subscribers) unmarshalMessage(data []byte) (ofs.SubscribeMessage, error) {
var err error
var message ofs.SubscribeMessage
if err = json.Unmarshal(data, &message); nil != err {
return message, err
}
if message.Message, err = message.MessageRaw.MarshalJSON(); nil != err {
return message, err
}
return message, nil
}
func (s *subscribers) Subscribe(h ofs.SubscriberHandler) error { func (s *subscribers) Subscribe(h ofs.SubscriberHandler) error {
if _, ok := s.subHandlers[h.GetChannel()]; ok { if _, ok := s.subHandlers[h.GetChannel()]; ok {
return ofs.ChannelExistError{Channel: h.GetChannel()} return ofs.ChannelExistError{Channel: h.GetChannel()}

5
subscribe_message.go
View File

@ -3,8 +3,9 @@ package overflow_subscriber
import "encoding/json" import "encoding/json"
type SubscribeMessage struct { type SubscribeMessage struct {
Targets []string `json:"targets"` Targets []string `json:"targets"`
Message *json.RawMessage `json:"message"` MessageRaw *json.RawMessage `json:"message"`
Message []byte
} }
type SubscribeMessageBody struct { type SubscribeMessageBody struct {