go-xmpp/websocket_transport.go

186 lines
4.4 KiB
Go
Raw Permalink Normal View History

2019-10-06 18:15:26 +00:00
package xmpp
import (
"bufio"
2019-10-06 18:15:26 +00:00
"context"
"encoding/xml"
2019-10-16 12:44:22 +00:00
"errors"
"fmt"
"io"
2019-10-06 18:15:26 +00:00
"strings"
"time"
"gosrc.io/xmpp/stanza"
2019-10-06 18:15:26 +00:00
"nhooyr.io/websocket"
)
const maxPacketSize = 32768
const pingTimeout = time.Duration(5) * time.Second
var ServerDoesNotSupportXmppOverWebsocket = errors.New("the websocket server does not support the xmpp subprotocol")
2019-10-16 12:44:22 +00:00
// The decoder is expected to be initialized after connecting to a server.
2019-10-06 18:15:26 +00:00
type WebsocketTransport struct {
Config TransportConfiguration
decoder *xml.Decoder
2019-10-06 18:15:26 +00:00
wsConn *websocket.Conn
queue chan []byte
logFile io.Writer
closeCtx context.Context
closeFunc context.CancelFunc
2019-10-06 18:15:26 +00:00
}
func (t *WebsocketTransport) Connect() (string, error) {
t.queue = make(chan []byte, 256)
t.closeCtx, t.closeFunc = context.WithCancel(context.Background())
2019-10-06 18:15:26 +00:00
var ctx context.Context
ctx = context.Background()
2019-10-12 15:50:00 +00:00
if t.Config.ConnectTimeout > 0 {
var cancelConnect context.CancelFunc
ctx, cancelConnect = context.WithTimeout(t.closeCtx, time.Duration(t.Config.ConnectTimeout)*time.Second)
defer cancelConnect()
2019-10-12 15:50:00 +00:00
}
2019-10-06 18:15:26 +00:00
wsConn, response, err := websocket.Dial(ctx, t.Config.Address, &websocket.DialOptions{
2019-10-16 12:44:22 +00:00
Subprotocols: []string{"xmpp"},
})
if err != nil {
return "", NewConnError(err, true)
2019-10-06 18:15:26 +00:00
}
2019-10-16 12:44:22 +00:00
if response.Header.Get("Sec-WebSocket-Protocol") != "xmpp" {
t.cleanup(websocket.StatusBadGateway)
return "", NewConnError(ServerDoesNotSupportXmppOverWebsocket, true)
2019-10-16 12:44:22 +00:00
}
wsConn.SetReadLimit(maxPacketSize)
t.wsConn = wsConn
t.startReader()
t.decoder = xml.NewDecoder(bufio.NewReaderSize(t, maxPacketSize))
t.decoder.CharsetReader = t.Config.CharsetReader
return t.StartStream()
}
func (t WebsocketTransport) StartStream() (string, error) {
if _, err := fmt.Fprintf(t, `<open xmlns="urn:ietf:params:xml:ns:xmpp-framing" to="%s" version="1.0" />`, t.Config.Domain); err != nil {
t.cleanup(websocket.StatusBadGateway)
return "", NewConnError(err, true)
}
sessionID, err := stanza.InitStream(t.GetDecoder())
if err != nil {
t.Close()
return "", NewConnError(err, false)
}
return sessionID, nil
2019-10-06 18:15:26 +00:00
}
// startReader runs a go function that keeps reading from the websocket. This
// is required to allow Ping() to work: Ping requires a Reader to be running
// to process incoming control frames.
func (t WebsocketTransport) startReader() {
go func() {
buffer := make([]byte, maxPacketSize)
for {
_, reader, err := t.wsConn.Reader(t.closeCtx)
if err != nil {
return
}
n, err := reader.Read(buffer)
if err != nil && err != io.EOF {
return
}
if n > 0 {
// We need to make a copy, otherwise we will overwrite the slice content
// on the next iteration of the for loop.
2019-10-21 08:05:43 +00:00
tmp := make([]byte, n)
copy(tmp, buffer)
t.queue <- tmp
}
}
}()
}
func (t WebsocketTransport) StartTLS() error {
return ErrTLSNotSupported
2019-10-11 05:15:47 +00:00
}
2019-10-06 18:15:26 +00:00
func (t WebsocketTransport) DoesStartTLS() bool {
return false
}
func (t WebsocketTransport) GetDomain() string {
return t.Config.Domain
}
func (t WebsocketTransport) GetDecoder() *xml.Decoder {
return t.decoder
}
2019-10-11 05:15:47 +00:00
func (t WebsocketTransport) IsSecure() bool {
return strings.HasPrefix(t.Config.Address, "wss:")
}
func (t WebsocketTransport) Ping() error {
ctx, cancel := context.WithTimeout(t.closeCtx, pingTimeout)
defer cancel()
return t.wsConn.Ping(ctx)
}
func (t *WebsocketTransport) Read(p []byte) (int, error) {
select {
case <-t.closeCtx.Done():
return 0, t.closeCtx.Err()
case data := <-t.queue:
if t.logFile != nil && len(data) > 0 {
_, _ = fmt.Fprintf(t.logFile, "RECV:\n%s\n\n", data)
}
copy(p, data)
return len(data), nil
}
2019-10-06 18:15:26 +00:00
}
func (t WebsocketTransport) Write(p []byte) (int, error) {
if t.logFile != nil {
_, _ = fmt.Fprintf(t.logFile, "SEND:\n%s\n\n", p)
}
return len(p), t.wsConn.Write(t.closeCtx, websocket.MessageText, p)
2019-10-06 18:15:26 +00:00
}
func (t WebsocketTransport) Close() error {
t.Write([]byte("<close xmlns=\"urn:ietf:params:xml:ns:xmpp-framing\" />"))
return t.cleanup(websocket.StatusGoingAway)
2019-10-06 18:15:26 +00:00
}
func (t *WebsocketTransport) LogTraffic(logFile io.Writer) {
t.logFile = logFile
}
func (t *WebsocketTransport) cleanup(code websocket.StatusCode) error {
var err error
if t.queue != nil {
close(t.queue)
t.queue = nil
}
if t.wsConn != nil {
err = t.wsConn.Close(websocket.StatusGoingAway, "Done")
t.wsConn = nil
}
if t.closeFunc != nil {
t.closeFunc()
t.closeFunc = nil
t.closeCtx = nil
}
return err
}
// ReceivedStreamClose is not used for websockets for now
func (t *WebsocketTransport) ReceivedStreamClose() {
return
}