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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
package gateway
import (
"fmt"
"sync"
"time"
"github.com/42wim/matterbridge/bridge"
"github.com/42wim/matterbridge/bridge/config"
"github.com/42wim/matterbridge/gateway/samechannel"
"github.com/sirupsen/logrus"
)
type Router struct {
config.Config
sync.RWMutex
BridgeMap map[string]bridge.Factory
Gateways map[string]*Gateway
Message chan config.Message
MattermostPlugin chan config.Message
logger *logrus.Entry
}
// NewRouter initializes a new Matterbridge router for the specified configuration and
// sets up all required gateways.
func NewRouter(rootLogger *logrus.Logger, cfg config.Config, bridgeMap map[string]bridge.Factory) (*Router, error) {
logger := rootLogger.WithFields(logrus.Fields{"prefix": "router"})
r := &Router{
Config: cfg,
BridgeMap: bridgeMap,
Message: make(chan config.Message),
MattermostPlugin: make(chan config.Message),
Gateways: make(map[string]*Gateway),
logger: logger,
}
sgw := samechannel.New(cfg)
gwconfigs := append(sgw.GetConfig(), cfg.BridgeValues().Gateway...)
for idx := range gwconfigs {
entry := &gwconfigs[idx]
if !entry.Enable {
continue
}
if entry.Name == "" {
return nil, fmt.Errorf("%s", "Gateway without name found")
}
if _, ok := r.Gateways[entry.Name]; ok {
return nil, fmt.Errorf("Gateway with name %s already exists", entry.Name)
}
r.Gateways[entry.Name] = New(rootLogger, entry, r)
}
return r, nil
}
// Start will connect all gateways belonging to this router and subsequently route messages
// between them.
func (r *Router) Start() error {
m := make(map[string]*bridge.Bridge)
for _, gw := range r.Gateways {
r.logger.Infof("Parsing gateway %s", gw.Name)
for _, br := range gw.Bridges {
m[br.Account] = br
}
}
for _, br := range m {
r.logger.Infof("Starting bridge: %s ", br.Account)
err := br.Connect()
if err != nil {
e := fmt.Errorf("Bridge %s failed to start: %v", br.Account, err)
if r.disableBridge(br, e) {
continue
}
return e
}
err = br.JoinChannels()
if err != nil {
e := fmt.Errorf("Bridge %s failed to join channel: %v", br.Account, err)
if r.disableBridge(br, e) {
continue
}
return e
}
}
// remove unused bridges
for _, gw := range r.Gateways {
for i, br := range gw.Bridges {
if br.Bridger == nil {
r.logger.Errorf("removing failed bridge %s", i)
delete(gw.Bridges, i)
}
}
}
go r.handleReceive()
//go r.updateChannelMembers()
return nil
}
// disableBridge returns true and empties a bridge if we have IgnoreFailureOnStart configured
// otherwise returns false
func (r *Router) disableBridge(br *bridge.Bridge, err error) bool {
if r.BridgeValues().General.IgnoreFailureOnStart {
r.logger.Error(err)
// setting this bridge empty
*br = bridge.Bridge{}
return true
}
return false
}
func (r *Router) getBridge(account string) *bridge.Bridge {
for _, gw := range r.Gateways {
if br, ok := gw.Bridges[account]; ok {
return br
}
}
return nil
}
func (r *Router) handleReceive() {
for msg := range r.Message {
msg := msg // scopelint
r.handleEventGetChannelMembers(&msg)
r.handleEventFailure(&msg)
r.handleEventRejoinChannels(&msg)
idx := 0
for _, gw := range r.Gateways {
// record all the message ID's of the different bridges
var msgIDs []*BrMsgID
if gw.ignoreMessage(&msg) {
continue
}
msg.Timestamp = time.Now()
gw.modifyMessage(&msg)
if idx == 0 {
gw.handleFiles(&msg)
}
for _, br := range gw.Bridges {
msgIDs = append(msgIDs, gw.handleMessage(&msg, br)...)
}
// only add the message ID if it doesn't already exists
if _, ok := gw.Messages.Get(msg.Protocol + " " + msg.ID); !ok && msg.ID != "" {
gw.Messages.Add(msg.Protocol+" "+msg.ID, msgIDs)
}
idx++
}
}
}
// updateChannelMembers sends every minute an GetChannelMembers event to all bridges.
func (r *Router) updateChannelMembers() {
// TODO sleep a minute because slack can take a while
// fix this by having actually connectionDone events send to the router
time.Sleep(time.Minute)
for {
for _, gw := range r.Gateways {
for _, br := range gw.Bridges {
// only for slack now
if br.Protocol != "slack" {
continue
}
r.logger.Debugf("sending %s to %s", config.EventGetChannelMembers, br.Account)
if _, err := br.Send(config.Message{Event: config.EventGetChannelMembers}); err != nil {
r.logger.Errorf("updateChannelMembers: %s", err)
}
}
}
time.Sleep(time.Minute)
}
}
|
