1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
|
// Copyright (c) 2015 Spinpunch, Inc. All Rights Reserved.
// See License.txt for license information.
package api
import (
l4g "code.google.com/p/log4go"
"github.com/mattermost/platform/model"
"github.com/mattermost/platform/store"
"strings"
)
type TeamHub struct {
connections map[*WebConn]bool
broadcast chan *model.Message
register chan *WebConn
unregister chan *WebConn
stop chan bool
teamId string
}
func NewTeamHub(teamId string) *TeamHub {
return &TeamHub{
broadcast: make(chan *model.Message),
register: make(chan *WebConn),
unregister: make(chan *WebConn),
connections: make(map[*WebConn]bool),
stop: make(chan bool),
teamId: teamId,
}
}
func (h *TeamHub) Register(webConn *WebConn) {
h.register <- webConn
}
func (h *TeamHub) Unregister(webConn *WebConn) {
h.unregister <- webConn
}
func (h *TeamHub) Stop() {
h.stop <- true
}
func (h *TeamHub) Start() {
pubsub := store.RedisClient().PubSub()
go func() {
defer func() {
l4g.Debug("redis reader finished for teamId=%v", h.teamId)
hub.Stop(h.teamId)
}()
l4g.Debug("redis reader starting for teamId=%v", h.teamId)
err := pubsub.Subscribe(h.teamId)
if err != nil {
l4g.Error("Error while subscribing to redis %v %v", h.teamId, err)
return
}
for {
if payload, err := pubsub.ReceiveTimeout(REDIS_WAIT); err != nil {
if strings.Contains(err.Error(), "i/o timeout") {
if len(h.connections) == 0 {
l4g.Debug("No active connections so sending stop %v", h.teamId)
return
}
} else {
return
}
} else {
msg := store.GetMessageFromPayload(payload)
if msg != nil {
h.broadcast <- msg
}
}
}
}()
go func() {
for {
select {
case webCon := <-h.register:
h.connections[webCon] = true
case webCon := <-h.unregister:
if _, ok := h.connections[webCon]; ok {
delete(h.connections, webCon)
close(webCon.Send)
}
case msg := <-h.broadcast:
for webCon := range h.connections {
if !(webCon.UserId == msg.UserId && msg.Action == model.ACTION_TYPING) {
select {
case webCon.Send <- msg:
default:
close(webCon.Send)
delete(h.connections, webCon)
}
}
}
case s := <-h.stop:
if s {
l4g.Debug("team hub stopping for teamId=%v", h.teamId)
for webCon := range h.connections {
webCon.WebSocket.Close()
}
pubsub.Close()
return
}
}
}
}()
}
|
