package xmpp
import (
"crypto/tls"
"fmt"
"strings"
"sync"
"time"
gxmpp "github.com/matterbridge/go-xmpp"
"github.com/rs/xid"
log "github.com/sirupsen/logrus"
. "git.deuxfleurs.fr/Deuxfleurs/easybridge/connector"
)
// User id format: username@server (= JID)
// OR: nickname@room_name@muc_server
// Room id format: room_name@muc_server (= MUC ID)
type XMPP struct {
handler Handler
connectorLoopNum int
connected bool
timeout int
server string
port int
ssl bool
jid string
jid_localpart string
password string
nickname string
conn *gxmpp.Client
stateLock sync.Mutex
muc map[RoomID]*mucInfo
}
type mucInfo struct {
joined bool
pendingJoins map[UserID]string
pendingLeaves map[UserID]struct{}
}
func (xm *XMPP) SetHandler(h Handler) {
xm.handler = h
}
func (xm *XMPP) Protocol() string {
return XMPP_PROTOCOL
}
func (xm *XMPP) Configure(c Configuration) error {
if xm.conn != nil {
xm.Close()
}
// Parse and validate configuration
var err error
xm.port, err = c.GetInt("port", 5222)
if err != nil {
return err
}
xm.ssl, err = c.GetBool("ssl", true)
if err != nil {
return err
}
xm.jid, err = c.GetString("jid")
if err != nil {
return err
}
jid_parts := strings.Split(xm.jid, "@")
if len(jid_parts) != 2 {
return fmt.Errorf("Invalid JID: %s", xm.jid)
}
xm.server = jid_parts[1]
xm.jid_localpart = jid_parts[0]
xm.nickname, _ = c.GetString("nickname", xm.jid_localpart)
xm.password, err = c.GetString("password")
if err != nil {
return err
}
// Try to connect
xm.muc = make(map[RoomID]*mucInfo)
xm.connectorLoopNum += 1
go xm.connectLoop(xm.connectorLoopNum)
for i := 0; i < 42; i++ {
time.Sleep(time.Duration(1) * time.Second)
if xm.connected {
return nil
}
}
return fmt.Errorf("Failed to connect after 42s attempting")
}
func (xm *XMPP) connectLoop(num int) {
xm.timeout = 10
for {
if xm.connectorLoopNum != num {
return
}
tc := &tls.Config{
ServerName: xm.server,
InsecureSkipVerify: true,
}
options := gxmpp.Options{
Host: xm.server,
User: xm.jid,
Password: xm.password,
NoTLS: true,
StartTLS: xm.ssl,
Session: true,
TLSConfig: tc,
}
var err error
xm.conn, err = options.NewClient()
if err != nil {
xm.connected = false
xm.handler.SystemMessage(fmt.Sprintf("XMPP failed to connect (%s). Retrying in %ds", err, xm.timeout))
time.Sleep(time.Duration(xm.timeout) * time.Second)
xm.timeout *= 2
if xm.timeout > 600 {
xm.timeout = 600
}
} else {
xm.connected = true
xm.timeout = 10
for muc, mucInfo := range xm.muc {
if mucInfo.joined {
_, err := xm.conn.JoinMUCNoHistory(string(muc), xm.nickname)
if err != nil {
xm.handler.SystemMessage(fmt.Sprintf("Could not rejoin MUC %s after reconnection: %s", muc, err))
xm.handler.Left(RoomID(muc))
mucInfo.joined = false
}
}
}
err = xm.handleXMPP()
xm.connected = false
xm.handler.SystemMessage(fmt.Sprintf("XMPP disconnected (%s), reconnecting)", err))
}
}
}
func (xm *XMPP) xmppKeepAlive() chan bool {
done := make(chan bool)
go func() {
ticker := time.NewTicker(90 * time.Second)
defer ticker.Stop()
for {
select {
case <-ticker.C:
if err := xm.conn.PingC2S("", ""); err != nil {
log.Debugf("PING failed %#v\n", err)
}
case <-done:
return
}
}
}()
return done
}
func (xm *XMPP) handleXMPP() error {
done := xm.xmppKeepAlive()
defer close(done)
for {
m, err := xm.conn.Recv()
if err != nil {
return err
}
log.Tracef("XMPP: %#v\n", m)
xm.handleXMPPStance(m)
}
}
func (xm *XMPP) handleXMPPStance(m interface{}) {
xm.stateLock.Lock()
defer xm.stateLock.Unlock()
switch v := m.(type) {
case gxmpp.Chat:
remote_sp := strings.Split(v.Remote, "/")
// Skip self-sent events
if v.Remote == xm.jid || (v.Type == "groupchat" && len(remote_sp) == 2 && remote_sp[1] == xm.nickname) {
return
}
// If empty text, make sure we joined the room
// We would do this at every incoming message if it were not so costly
if v.Text == "" && v.Type == "groupchat" {
xm.handler.Joined(RoomID(remote_sp[0]))
}
// Handle subject change in group chats
if v.Subject != "" && v.Type == "groupchat" {
author := UserID("")
if len(remote_sp) == 2 {
if remote_sp[1] == xm.nickname {
author = xm.User()
} else {
author = UserID(remote_sp[1] + "@" + remote_sp[0])
}
}
xm.handler.RoomInfoUpdated(RoomID(remote_sp[0]), author, &RoomInfo{
Topic: v.Subject,
})
}
// Handle text message
if v.Text != "" {
event := &Event{
Type: EVENT_MESSAGE,
Text: v.Text,
}
if strings.HasPrefix(event.Text, "/me ") {
event.Type = EVENT_ACTION
event.Text = strings.Replace(event.Text, "/me ", "", 1)
}
if v.Type == "chat" {
event.Author = UserID(remote_sp[0])
xm.handler.Event(event)
}
if v.Type == "groupchat" && len(remote_sp) == 2 {
// First flush pending leaves and joins
room_id := RoomID(remote_sp[0])
if muc, ok := xm.muc[room_id]; ok {
muc.flushLeavesJoins(room_id, xm.handler)
}
// Now send event
event.Room = room_id
event.Author = UserID(remote_sp[1] + "@" + remote_sp[0])
event.Id = v.ID
xm.handler.Event(event)
}
}
case gxmpp.Presence:
remote := strings.Split(v.From, "/")
room := RoomID(remote[0])
if mucInfo, ok := xm.muc[room]; ok {
// skip presence with no user and self-presence
if len(remote) < 2 || remote[1] == xm.nickname {
return
}
user := UserID(remote[1] + "@" + remote[0])
if v.Type != "unavailable" {
if _, ok := mucInfo.pendingLeaves[user]; ok {
delete(mucInfo.pendingLeaves, user)
} else {
mucInfo.pendingJoins[user] = remote[1]
}
} else {
if _, ok := mucInfo.pendingJoins[user]; ok {
delete(mucInfo.pendingJoins, user)
} else {
mucInfo.pendingLeaves[user] = struct{}{}
}
}
}
}
}
func (muc *mucInfo) flushLeavesJoins(room RoomID, handler Handler) {
for user, display_name := range muc.pendingJoins {
handler.Event(&Event{
Type: EVENT_JOIN,
Room: room,
Author: user,
})
handler.UserInfoUpdated(user, &UserInfo{
DisplayName: display_name,
})
}
for user, _ := range muc.pendingLeaves {
handler.Event(&Event{
Type: EVENT_LEAVE,
Room: room,
Author: user,
})
}
muc.pendingJoins = make(map[UserID]string)
muc.pendingLeaves = make(map[UserID]struct{})
}
func (xm *XMPP) User() UserID {
return UserID(xm.jid)
}
func (xm *XMPP) SetUserInfo(info *UserInfo) error {
return fmt.Errorf("Not implemented")
}
func (xm *XMPP) SetRoomInfo(roomId RoomID, info *RoomInfo) error {
if info.Topic != "" {
_, err := xm.conn.Send(gxmpp.Chat{
Type: "groupchat",
Remote: string(roomId),
Subject: info.Topic,
})
if err != nil {
return err
}
}
if info.Picture.MediaObject != nil {
// TODO
return fmt.Errorf("Room picture change not implemented on xmpp")
}
if info.Name != "" && info.Name != string(roomId) {
// TODO
return fmt.Errorf("Room name change not implemented on xmpp")
}
return nil
}
func (xm *XMPP) Join(roomId RoomID) error {
xm.stateLock.Lock()
defer xm.stateLock.Unlock()
xm.muc[roomId] = &mucInfo{
pendingJoins: make(map[UserID]string),
pendingLeaves: make(map[UserID]struct{}),
}
log.Tracef("Join %s with nick %s\n", roomId, xm.nickname)
_, err := xm.conn.JoinMUCNoHistory(string(roomId), xm.nickname)
if err == nil {
xm.muc[roomId].joined = true
}
return err
}
func (xm *XMPP) Invite(userId UserID, roomId RoomID) error {
if roomId == "" {
xm.conn.RequestSubscription(string(userId))
xm.conn.ApproveSubscription(string(userId))
return nil
}
// TODO
return fmt.Errorf("Not implemented")
}
func (xm *XMPP) Leave(roomId RoomID) {
xm.stateLock.Lock()
defer xm.stateLock.Unlock()
xm.conn.LeaveMUC(string(roomId))
if muc, ok := xm.muc[roomId]; ok {
muc.joined = false
}
}
func (xm *XMPP) SearchForUsers(query string) ([]UserSearchResult, error) {
// TODO: search roster
return nil, fmt.Errorf("Not implemented")
}
func (xm *XMPP) Send(event *Event) (string, error) {
xm.stateLock.Lock()
defer xm.stateLock.Unlock()
if event.Attachments != nil && len(event.Attachments) > 0 {
for _, at := range event.Attachments {
url := at.URL()
if url == "" {
// TODO find a way to send them using some hosing of some kind
return "", fmt.Errorf("Attachment without URL sent to XMPP")
} else {
event.Text += fmt.Sprintf("\n%s (%s, %dkb)",
url, at.Mimetype(), at.Size()/1024)
}
}
}
if event.Id == "" {
event.Id = xid.New().String()
}
log.Tracef("xm *XMPP Send %#v\n", event)
if len(event.Recipient) > 0 {
_, err := xm.conn.Send(gxmpp.Chat{
Type: "chat",
Remote: string(event.Recipient),
Text: event.Text,
})
return event.Id, err
} else if len(event.Room) > 0 {
if muc, ok := xm.muc[event.Room]; ok {
muc.flushLeavesJoins(event.Room, xm.handler)
}
_, err := xm.conn.Send(gxmpp.Chat{
Type: "groupchat",
Remote: string(event.Room),
Text: event.Text,
ID: event.Id,
})
return event.Id, err
} else {
return "", fmt.Errorf("Invalid event")
}
}
func (xm *XMPP) UserCommand(cmd string) {
xm.handler.SystemMessage("Command not supported.")
}
func (xm *XMPP) Close() {
xm.stateLock.Lock()
defer xm.stateLock.Unlock()
if xm.conn != nil {
xm.conn.Close()
}
xm.conn = nil
xm.connectorLoopNum += 1
xm.connected = false
}
