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
|
package xmpp
import (
"github.com/pkg/errors"
"time"
"dev.narayana.im/narayana/telegabber/config"
"dev.narayana.im/narayana/telegabber/persistence"
"dev.narayana.im/narayana/telegabber/telegram"
"dev.narayana.im/narayana/telegabber/xmpp/gateway"
log "github.com/sirupsen/logrus"
"gosrc.io/xmpp"
)
var tgConf config.TelegramConfig
var sessions map[string]*telegram.Client
var db *persistence.SessionsYamlDB
// NewComponent starts a new component and wraps it in
// a stream manager that you should start yourself
func NewComponent(conf config.XMPPConfig, tc config.TelegramConfig) (*xmpp.StreamManager, *xmpp.Component, error) {
var err error
gateway.Jid, err = xmpp.NewJid(conf.Jid)
if err != nil {
return nil, nil, err
}
tgConf = tc
options := xmpp.ComponentOptions{
Address: conf.Host + ":" + conf.Port,
Domain: conf.Jid,
Secret: conf.Password,
Name: "telegabber",
}
router := xmpp.NewRouter()
router.HandleFunc("iq", HandleIq)
router.HandleFunc("presence", HandlePresence)
router.HandleFunc("message", HandleMessage)
component, err := xmpp.NewComponent(options, router)
if err != nil {
return nil, nil, err
}
// probe all known sessions
err = loadSessions(conf.Db, component)
if err != nil {
return nil, nil, err
}
sm := xmpp.NewStreamManager(component, func(s xmpp.Sender) {
go heartbeat(component)
})
return sm, component, nil
}
func heartbeat(component *xmpp.Component) {
var err error
probeType := gateway.SPType("probe")
for jid := range sessions {
err = gateway.SendPresence(component, jid, probeType)
if err != nil {
log.Error(err)
}
}
log.Info("Starting heartbeat queue")
// status updater thread
for {
time.Sleep(60e9)
for key, presence := range gateway.Queue {
err = gateway.ResumableSend(component, presence)
if err != nil {
gateway.LogBadPresence(presence)
} else {
delete(gateway.Queue, key)
}
}
}
}
func loadSessions(dbPath string, component *xmpp.Component) error {
var err error
sessions = make(map[string]*telegram.Client)
db, err = persistence.LoadSessions(dbPath)
if err != nil {
return err
}
db.Transaction(func() bool {
for jid, session := range db.Data.Sessions {
getTelegramInstance(jid, &session, component)
}
return false
}, persistence.SessionMarshaller)
return nil
}
func getTelegramInstance(jid string, savedSession *persistence.Session, component *xmpp.Component) (*telegram.Client, bool) {
var err error
session, ok := sessions[jid]
if !ok {
session, err = telegram.NewClient(tgConf, jid, component, savedSession)
if err != nil {
log.Error(errors.Wrap(err, "TDlib initialization failure"))
return session, false
}
sessions[jid] = session
}
return session, true
}
// Close gracefully terminates the component and saves active sessions
func Close(component *xmpp.Component) {
log.Error("Disconnecting...")
// close all sessions
for _, session := range sessions {
session.Disconnect()
}
// save sessions
db.Transaction(func() bool {
for jid, session := range sessions {
db.Data.Sessions[jid] = *session.Session
}
return true
}, persistence.SessionMarshaller)
// close stream
component.Disconnect()
}
|
