easybridge/connector/xmpp/xmpp.go

package xmpp

import (
	"time"
	//"os"
	"strings"
	"fmt"
	"crypto/tls"
	"encoding/xml"

	log "github.com/sirupsen/logrus"
	//gxmpp "github.com/mattn/go-xmpp"
	gxmpp "git.deuxfleurs.fr/lx/go-xmpp"

	. "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

	isMUC map[string]bool
}

func (xm *XMPP) SetHandler(h Handler) {
	xm.handler = h
}

func(xm *XMPP) Protocol() string {
	return "xmpp"
}

func (xm *XMPP) Configure(c Configuration) error {
	if xm.conn != nil {
		xm.Close()
	}

	// Parse and validate configuration
	var err error

	xm.server, err = c.GetString("server")
	if err != nil {
		return err
	}

	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)
	}
	if jid_parts[1] != xm.server {
		return fmt.Errorf("JID %s not on server %s", xm.jid, xm.server)
	}
	xm.jid_localpart = jid_parts[0]
	xm.nickname = xm.jid_localpart

	xm.password, err = c.GetString("password")
	if err != nil {
		return err
	}

	// Try to connect
	if xm.isMUC == nil {
		xm.isMUC = make(map[string]bool)
	}

	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:         strings.Split(xm.jid, "@")[1],
			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
			fmt.Printf("XMPP failed to connect / disconnected: %s\n", err)
			fmt.Printf("Retrying in %ds\n", 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
			err = xm.handleXMPP()
			if err != nil {
				xm.connected = false
				fmt.Printf("XMPP disconnected: %s\n", err)
				fmt.Printf("Reconnecting.\n")
			}
		}
	}
}

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.Printf("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
		}


		switch v := m.(type) {
		case gxmpp.Chat:
			fmt.Printf("XMPP chat: %#v\n", v)
			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) {
				continue
			}

			// 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 {
					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 {
					event.Room = RoomID(remote_sp[0])
					event.Author = UserID(remote_sp[1] + "@" + remote_sp[0])
					xm.handler.Event(event)
				}
			}
		case gxmpp.Presence:
			fmt.Printf("XMPP presence: %#v\n", v)

			remote := strings.Split(v.From, "/")
			if ismuc, ok := xm.isMUC[remote[0]]; ok && ismuc {
				// skip presence with no user and self-presence
				if len(remote) < 2 || remote[1] == xm.nickname {
					continue
				}

				user := UserID(remote[1] + "@" + remote[0])
				event := &Event{
					Type: EVENT_JOIN,
					Room: RoomID(remote[0]),
					Author: user,
				}
				if v.Type == "unavailable" {
					event.Type = EVENT_LEAVE
				}
				xm.handler.Event(event)
				xm.handler.UserInfoUpdated(user, &UserInfo{
					DisplayName: remote[1],
				})
			} else {
				// Send discovery query
				iq, err := xml.Marshal(&DiscoQuery{})
				if err != nil {
					fmt.Printf("XML marshall error: %s\n", err)
				} else {
					xm.conn.SendIQ(gxmpp.IQ{
						Type: "get",
						To: remote[0],
						ID: "items1",
						Query: iq,
					})
				}
			}
		case gxmpp.IQ:
			fmt.Printf("XMPP iq: from=%s to=%s id=%s type=%s\n", v.From, v.To, v.ID, v.Type)
			if len(v.Query) > 0 {
				fmt.Printf("Query data: %s\n", string(v.Query))
			}

			if v.Type == "result" && v.ID == "items1" {
				var q DiscoQuery
				err := xml.Unmarshal(v.Query, &q)
				if err != nil {
					fmt.Printf("XML unmarshall error: %s\n", err)
					continue
				}

				for _, item := range q.Items {
					if item.Node == "urn:xmpp:avatar:metadata" || item.Node == "urn:xmpp:avatar:data" {
						sub := &PubSub{
							Subscribe: []Subscribe{
								Subscribe{
									Jid: xm.jid,
									Node: item.Node,
								},
							},
						}
						iq, err := xml.Marshal(sub)
						if err != nil {
							fmt.Printf("XML marshall error: %s\n", err)
						} else {
							fmt.Printf("IQ AVATAR SUB: %s\n", iq)
							xm.conn.SendIQ(gxmpp.IQ{
								To: v.From,
								Type: "set",
								ID: "sub1",
								Query: iq,
							})
						}
					}
				}
			}
		}
	}
}

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 != "" {
		xm.conn.Send(gxmpp.Chat{
			Type: "groupchat",
			Remote: string(roomId),
			Subject: info.Topic,
		})
	}

	if info.Picture != 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.isMUC[string(roomId)] = true

	fmt.Printf("Join %s with nick %s\n", roomId, xm.nickname)
	_, err := xm.conn.JoinMUCNoHistory(string(roomId), xm.nickname)
	return err
}

func (xm *XMPP) Invite(userId UserID, roomId RoomID) error {
	// TODO
	return fmt.Errorf("Not implemented")
}

func (xm *XMPP) Leave(roomId RoomID) {
	xm.conn.LeaveMUC(string(roomId))
}

func (xm *XMPP) Send(event *Event) error {
	fmt.Printf("xm *XMPP Send %#v\n", event)
	if len(event.Recipient) > 0 {
		xm.conn.Send(gxmpp.Chat{
			Type: "chat",
			Remote: string(event.Recipient),
			Text: event.Text,
		})
		return nil
	} else if len(event.Room) > 0 {
		xm.conn.Send(gxmpp.Chat{
			Type: "groupchat",
			Remote: string(event.Room),
			Text: event.Text,
		})
		return nil
	} else {
		return fmt.Errorf("Invalid event")
	}
}

func (xm *XMPP) Close() {
	xm.conn.Close()
	xm.conn = nil
	xm.connectorLoopNum += 1
}