Repository: FluuxIO/go-xmpp (go)

Issue:
Make client_manager able to manage any connection
Client manager should be able to handle reconnect and exponential back-off on both client connection and component connection.

Pull request:
Turn ClientManager into StreamManager
The StreamManager can supervise and handle events for both an XMPP client or an XMPP component.

Fix #38 

Patch:
diff --git a/_examples/go.mod b/_examples/go.mod
index 352eef6..84f9745 100644
--- a/_examples/go.mod
+++ b/_examples/go.mod
@@ -2,4 +2,10 @@ module gosrc.io/xmpp/_examples
 
 go 1.12
 
-require gosrc.io/xmpp v0.0.0-20190608091551-b7461ae97fed
+require (
+	github.com/processone/mpg123 v1.0.0
+	github.com/processone/soundcloud v1.0.0
+	gosrc.io/xmpp v0.0.0-20190608160922-63a29d5c218a
+)
+
+replace gosrc.io/xmpp => gosrc.io/xmpp v0.0.0-20190608160922-63a29d5c218a
diff --git a/_examples/go.sum b/_examples/go.sum
index 943df1c..123436d 100644
--- a/_examples/go.sum
+++ b/_examples/go.sum
@@ -1,11 +1,8 @@
 github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M=
-github.com/processone/mpg123 v1.0.0 h1:o2WOyGZRM255or1Zc/LtF/jARn51B+9aQl72Qace0GA=
 github.com/processone/mpg123 v1.0.0/go.mod h1:X/FeL+h8vD1bYsG9tIWV3M2c4qNTZOficyvPVBP08go=
-github.com/processone/soundcloud v1.0.0 h1:/+i6+Yveb7Y6IFGDSkesYI+HddblzcRTQClazzVHxoE=
 github.com/processone/soundcloud v1.0.0/go.mod h1:kDLeWpkRtN3C8kIReQdxoiRi92P9xR6yW6qLOJnNWfY=
-golang.org/x/net v0.0.0-20190110200230-915654e7eabc h1:Yx9JGxI1SBhVLFjpAkWMaO1TF+xyqtHLjZpvQboJGiM=
 golang.org/x/net v0.0.0-20190110200230-915654e7eabc/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522 h1:bhOzK9QyoD0ogCnFro1m2mz41+Ib0oOhfJnBp5MR4K4=
 golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
-gosrc.io/xmpp v0.0.0-20190608091551-b7461ae97fed h1:uLrripMKk85UZ0Kd9V2S7qQy2CM4kveMZkCBqBEOqmY=
-gosrc.io/xmpp v0.0.0-20190608091551-b7461ae97fed/go.mod h1:it3z4S42Sy7eHWFqwmdFJbygg7lCmbrhKeqK7HQSqSU=
+gosrc.io/xmpp v0.0.0-20190608160922-63a29d5c218a h1:TKUhPFlQkBLpoSTNgh4rX64T1FmObGcggIYl7d2q4vM=
+gosrc.io/xmpp v0.0.0-20190608160922-63a29d5c218a/go.mod h1:6NJG4vRCxQJMGLxIdroPLPd++FPLOmDqJdJEt2mu4kQ=
diff --git a/_examples/xmpp_component/xmpp_component.go b/_examples/xmpp_component/xmpp_component.go
index 4b76a82..7b9174c 100644
--- a/_examples/xmpp_component/xmpp_component.go
+++ b/_examples/xmpp_component/xmpp_component.go
@@ -8,31 +8,41 @@ import (
 )
 
 func main() {
-	component := MyComponent{Name: "Test Component", Category: "gateway", Type: "service"}
-	component.xmpp = &xmpp.Component{Host: "service.localhost", Secret: "mypass"}
-	if err := component.xmpp.Connect("localhost:8888"); err != nil {
-		log.Fatal(err)
+	opts := xmpp.ComponentOptions{
+		Domain:   "service.localhost",
+		Secret:   "mypass",
+		Address:  "localhost:8888",
+		Name:     "Test Component",
+		Category: "gateway",
+		Type:     "service",
+	}
+	component, err := xmpp.NewComponent(opts)
+	if err != nil {
+		log.Fatalf("%+v", err)
 	}
 
-	for {
-		packet, err := component.xmpp.ReadPacket()
-		if err != nil {
-			fmt.Println("read error", err)
-			return
-		}
+	// If you pass the component to a connection manager, it will handle the reconnect policy
+	// for you automatically.
+	cm := xmpp.NewStreamManager(component, nil)
+	err = cm.Start()
+	if err != nil {
+		log.Fatal(err)
+	}
 
+	// Iterator to receive packets coming from our XMPP connection
+	for packet := range component.Recv() {
 		switch p := packet.(type) {
 		case xmpp.IQ:
 			switch inner := p.Payload[0].(type) {
 			case *xmpp.DiscoInfo:
-				fmt.Println("Disco Info")
+				fmt.Println("DiscoInfo")
 				if p.Type == "get" {
-					DiscoResult(component, p.PacketAttrs, inner)
+					discoResult(component, p.PacketAttrs, inner)
 				}
 			case *xmpp.DiscoItems:
 				fmt.Println("DiscoItems")
 				if p.Type == "get" {
-					DiscoItems(component, p.PacketAttrs, inner)
+					discoItems(component, p.PacketAttrs, inner)
 				}
 			default:
 				fmt.Println("ignoring iq packet", inner)
@@ -42,7 +52,7 @@ func main() {
 					Type:   "cancel",
 				}
 				reply := p.MakeError(xError)
-				_ = component.xmpp.Send(&reply)
+				_ = component.Send(&reply)
 			}
 
 		case xmpp.Message:
@@ -57,16 +67,7 @@ func main() {
 	}
 }
 
-type MyComponent struct {
-	Name string
-	// Typical categories and types: https://xmpp.org/registrar/disco-categories.html
-	Category string
-	Type     string
-
-	xmpp *xmpp.Component
-}
-
-func DiscoResult(c MyComponent, attrs xmpp.PacketAttrs, info *xmpp.DiscoInfo) {
+func discoResult(c *xmpp.Component, attrs xmpp.PacketAttrs, info *xmpp.DiscoInfo) {
 	iq := xmpp.NewIQ("result", attrs.To, attrs.From, attrs.Id, "en")
 	var identity xmpp.Identity
 	if info.Node == "" {
@@ -86,10 +87,10 @@ func DiscoResult(c MyComponent, attrs xmpp.PacketAttrs, info *xmpp.DiscoInfo) {
 	}
 	iq.AddPayload(&payload)
 
-	_ = c.xmpp.Send(iq)
+	_ = c.Send(iq)
 }
 
-func DiscoItems(c MyComponent, attrs xmpp.PacketAttrs, items *xmpp.DiscoItems) {
+func discoItems(c *xmpp.Component, attrs xmpp.PacketAttrs, items *xmpp.DiscoItems) {
 	iq := xmpp.NewIQ("result", attrs.To, attrs.From, attrs.Id, "en")
 
 	var payload xmpp.DiscoItems
@@ -101,5 +102,5 @@ func DiscoItems(c MyComponent, attrs xmpp.PacketAttrs, items *xmpp.DiscoItems) {
 		}
 	}
 	iq.AddPayload(&payload)
-	_ = c.xmpp.Send(iq)
+	_ = c.Send(iq)
 }
diff --git a/_examples/xmpp_echo/xmpp_echo.go b/_examples/xmpp_echo/xmpp_echo.go
index 0677e0c..f87cdab 100644
--- a/_examples/xmpp_echo/xmpp_echo.go
+++ b/_examples/xmpp_echo/xmpp_echo.go
@@ -28,7 +28,7 @@ func main() {
 
 	// If you pass the client to a connection manager, it will handle the reconnect policy
 	// for you automatically.
-	cm := xmpp.NewClientManager(client, nil)
+	cm := xmpp.NewStreamManager(client, nil)
 	err = cm.Start()
 	if err != nil {
 		log.Fatal(err)
diff --git a/_examples/xmpp_jukebox/xmpp_jukebox.go b/_examples/xmpp_jukebox/xmpp_jukebox.go
index b020399..f5e373f 100644
--- a/_examples/xmpp_jukebox/xmpp_jukebox.go
+++ b/_examples/xmpp_jukebox/xmpp_jukebox.go
@@ -107,7 +107,7 @@ func connectXmpp(jid string, password string, address string) (client *xmpp.Clie
 		return
 	}
 
-	if _, err = client.Connect(); err != nil {
+	if err = client.Connect(); err != nil {
 		return
 	}
 	return
diff --git a/client.go b/client.go
index c3a2802..324e297 100644
--- a/client.go
+++ b/client.go
@@ -11,6 +11,7 @@ import (
 )
 
 //=============================================================================
+// EventManager
 
 // ConnState represents the current connection state.
 type ConnState = uint8
@@ -108,7 +109,7 @@ func NewClient(config Config) (c *Client, err error) {
 		c.config.ConnectTimeout = 15 // 15 second as default
 	}
 
-	// Create a default channel that developer can override
+	// Create a default channel that developers can override
 	c.RecvChannel = make(chan interface{})
 
 	return
@@ -133,18 +134,18 @@ func checkAddress(addr string) (string, error) {
 }
 
 // Connect triggers actual TCP connection, based on previously defined parameters.
-func (c *Client) Connect() (*Session, error) {
+func (c *Client) Connect() error {
 	var err error
 
 	c.conn, err = net.DialTimeout("tcp", c.config.Address, time.Duration(c.config.ConnectTimeout)*time.Second)
 	if err != nil {
-		return nil, err
+		return err
 	}
 	c.updateState(StateConnected)
 
-	// Connection is ok, we now open XMPP session
+	// Client is ok, we now open XMPP session
 	if c.conn, c.Session, err = NewSession(c.conn, c.config); err != nil {
-		return c.Session, err
+		return err
 	}
 	c.updateState(StateSessionEstablished)
 
@@ -157,7 +158,7 @@ func (c *Client) Connect() (*Session, error) {
 	// Start the receiver go routine
 	go c.recv()
 
-	return c.Session, err
+	return err
 }
 
 func (c *Client) Disconnect() {
@@ -166,6 +167,17 @@ func (c *Client) Disconnect() {
 	_ = c.conn.Close()
 }
 
+func (c *Client) SetHandler(handler EventHandler) {
+	c.Handler = handler
+}
+
+// Recv abstracts receiving preparsed XMPP packets from a channel.
+// Channel allow client to receive / dispatch packets in for range loop.
+// TODO: Deprecate this function in favor of reading directly from the RecvChannel ?
+func (c *Client) Recv() <-chan interface{} {
+	return c.RecvChannel
+}
+
 func (c *Client) recv() (err error) {
 	for {
 		val, err := next(c.Session.decoder)
@@ -187,13 +199,6 @@ func (c *Client) recv() (err error) {
 	}
 }
 
-// Recv abstracts receiving preparsed XMPP packets from a channel.
-// Channel allow client to receive / dispatch packets in for range loop.
-// TODO: Deprecate this function in favor of reading directly from the RecvChannel
-func (c *Client) Recv() <-chan interface{} {
-	return c.RecvChannel
-}
-
 // Send marshalls XMPP stanza and sends it to the server.
 func (c *Client) Send(packet Packet) error {
 	data, err := xml.Marshal(packet)
diff --git a/component.go b/component.go
index 67c4bfd..966d4ed 100644
--- a/component.go
+++ b/component.go
@@ -13,12 +13,41 @@ import (
 
 const componentStreamOpen = "<?xml version='1.0'?><stream:stream to='%s' xmlns='%s' xmlns:stream='%s'>"
 
+type ComponentOptions struct {
+	// =================================
+	// Component Connection Info
+
+	// Domain is the XMPP server subdomain that the component will handle
+	Domain string
+	// Secret is the "password" used by the XMPP server to secure component access
+	Secret string
+	// Address is the XMPP Host and port to connect to. Host is of
+	// the form 'serverhost:port' i.e "localhost:8888"
+	Address string
+
+	// =================================
+	// Component discovery
+
+	// Component human readable name, that will be shown in XMPP discovery
+	Name string
+	// Typical categories and types: https://xmpp.org/registrar/disco-categories.html
+	Category string
+	Type     string
+
+	// =================================
+	// Communication with developer client / StreamManager
+
+	// Packet channel
+	RecvChannel chan interface{}
+	// Track and broadcast connection state
+	EventManager
+}
+
 // Component implements an XMPP extension allowing to extend XMPP server
 // using external components. Component specifications are defined
 // in XEP-0114, XEP-0355 and XEP-0356.
 type Component struct {
-	Host   string
-	Secret string
+	ComponentOptions
 
 	// TCP level connection
 	conn net.Conn
@@ -28,18 +57,25 @@ type Component struct {
 	decoder     *xml.Decoder
 }
 
+func NewComponent(opts ComponentOptions) (*Component, error) {
+	c := Component{ComponentOptions: opts}
+	// Create a default channel that developers can override
+	c.RecvChannel = make(chan interface{})
+	return &c, nil
+}
+
 // Connect triggers component connection to XMPP server component port.
-// TODO Helper to prepare connection string
-func (c *Component) Connect(connStr string) error {
+// TODO: Failed handshake should be a permanent error
+func (c *Component) Connect() error {
 	var conn net.Conn
 	var err error
-	if conn, err = net.DialTimeout("tcp", connStr, time.Duration(5)*time.Second); err != nil {
+	if conn, err = net.DialTimeout("tcp", c.Address, time.Duration(5)*time.Second); err != nil {
 		return err
 	}
 	c.conn = conn
 
 	// 1. Send stream open tag
-	if _, err := fmt.Fprintf(conn, componentStreamOpen, c.Host, NSComponent, NSStream); err != nil {
+	if _, err := fmt.Fprintf(conn, componentStreamOpen, c.Domain, NSComponent, NSStream); err != nil {
 		return errors.New("cannot send stream open " + err.Error())
 	}
 	c.decoder = xml.NewDecoder(conn)
@@ -65,17 +101,49 @@ func (c *Component) Connect(connStr string) error {
 	case StreamError:
 		return errors.New("handshake failed " + v.Error.Local)
 	case Handshake:
+		// Start the receiver go routine
+		go c.recv()
 		return nil
 	default:
-		return errors.New("unexpected packet, got " + v.Name())
+		return errors.New("expecting handshake result, got " + v.Name())
 	}
-	panic("unreachable")
 }
 
-// ReadPacket reads next incoming XMPP packet
-func (c *Component) ReadPacket() (Packet, error) {
-	// TODO use defined interface Packet
-	return next(c.decoder)
+func (c *Component) Disconnect() {
+	_ = c.SendRaw("</stream:stream>")
+	// TODO: Add a way to wait for stream close acknowledgement from the server for clean disconnect
+	_ = c.conn.Close()
+}
+
+func (c *Component) SetHandler(handler EventHandler) {
+	c.Handler = handler
+}
+
+// Recv abstracts receiving preparsed XMPP packets from a channel.
+// Channel allow client to receive / dispatch packets in for range loop.
+// TODO: Deprecate this function in favor of reading directly from the RecvChannel ?
+func (c *Component) Recv() <-chan interface{} {
+	return c.RecvChannel
+}
+
+func (c *Component) recv() (err error) {
+	for {
+		val, err := next(c.decoder)
+		if err != nil {
+			c.updateState(StateDisconnected)
+			return err
+		}
+
+		// Handle stream errors
+		switch p := val.(type) {
+		case StreamError:
+			c.RecvChannel <- val
+			close(c.RecvChannel)
+			c.streamError(p.Error.Local, p.Text)
+			return errors.New("stream error: " + p.Error.Local)
+		}
+		c.RecvChannel <- val
+	}
 }
 
 // Send marshalls XMPP stanza and sends it to the server.
@@ -96,8 +164,9 @@ func (c *Component) Send(packet Packet) error {
 // disconnect the component. It is up to the user of this method to
 // carefully craft the XML content to produce valid XMPP.
 func (c *Component) SendRaw(packet string) error {
-	fmt.Fprintf(c.conn, packet) // TODO handle errors
-	return nil
+	var err error
+	_, err = fmt.Fprintf(c.conn, packet)
+	return err
 }
 
 // handshake generates an authentication token based on StreamID and shared secret.
@@ -143,3 +212,34 @@ func (handshakeDecoder) decode(p *xml.Decoder, se xml.StartElement) (Handshake,
 	err := p.DecodeElement(&packet, &se)
 	return packet, err
 }
+
+// Service discovery
+
+func (c *Component) discoResult(attrs PacketAttrs, info *DiscoInfo) {
+	iq := NewIQ("result", attrs.To, attrs.From, attrs.Id, "en")
+	var identity Identity
+	if info.Node == "" {
+		identity = Identity{
+			Name:     c.Name,
+			Category: c.Category,
+			Type:     c.Type,
+		}
+	}
+
+	payload := DiscoInfo{
+		Identity: identity,
+		Features: []Feature{
+			{Var: NSDiscoInfo},
+			{Var: NSDiscoItems},
+		},
+	}
+	iq.AddPayload(&payload)
+
+	_ = c.Send(iq)
+}
+
+/*
+TODO: Add support for discovery management directly in component
+TODO: Support multiple identities on disco info
+TODO: Support returning features on disco info
+*/
diff --git a/config.go b/config.go
index 79770b1..92748ac 100644
--- a/config.go
+++ b/config.go
@@ -12,7 +12,7 @@ type Config struct {
 	Password       string
 	PacketLogger   *os.File // Used for debugging
 	Lang           string   // TODO: should default to 'en'
-	ConnectTimeout int      // Connection timeout in seconds. Default to 15
+	ConnectTimeout int      // Client timeout in seconds. Default to 15
 	// Insecure can be set to true to allow to open a session without TLS. If TLS
 	// is supported on the server, we will still try to use it.
 	Insecure      bool
diff --git a/client_manager.go b/stream_manager.go
similarity index 60%
rename from client_manager.go
rename to stream_manager.go
index 75599be..b353c2b 100644
--- a/client_manager.go
+++ b/stream_manager.go
@@ -1,74 +1,97 @@
 package xmpp // import "gosrc.io/xmpp"
 
 import (
+	"errors"
 	"time"
 
 	"golang.org/x/xerrors"
 )
 
-type PostConnect func(c *Client)
+// The Fluux XMPP lib can manage client or component XMPP streams.
+// The StreamManager handles the stream workflow handling the common
+// stream events and doing the right operations.
+//
+// It can handle:
+//     - Client
+//     - Stream establishment workflow
+//     - Reconnection strategies, with exponential backoff. It also takes into account
+//       permanent errors to avoid useless reconnection loops.
+//     - Metrics processing
+
+type StreamClient interface {
+	Connect() error
+	Disconnect()
+	SetHandler(handler EventHandler)
+}
 
-// ClientManager supervises an XMPP client connection. Its role is to handle connection events and
+// StreamManager supervises an XMPP client connection. Its role is to handle connection events and
 // apply reconnection strategy.
-type ClientManager struct {
-	Client      *Client
-	Session     *Session
+type StreamManager struct {
+	client      StreamClient
 	PostConnect PostConnect
 
 	// Store low level metrics
 	Metrics *Metrics
 }
 
-// NewClientManager creates a new client manager structure, intended to support
+type PostConnect func(c StreamClient)
+
+// NewStreamManager creates a new StreamManager structure, intended to support
 // handling XMPP client state event changes and auto-trigger reconnection
-// based on ClientManager configuration.
-func NewClientManager(client *Client, pc PostConnect) *ClientManager {
-	return &ClientManager{
-		Client:      client,
+// based on StreamManager configuration.
+// TODO: Move parameters to Start and remove factory method
+func NewStreamManager(client StreamClient, pc PostConnect) *StreamManager {
+	return &StreamManager{
+		client:      client,
 		PostConnect: pc,
 	}
 }
 
 // Start launch the connection loop
-func (cm *ClientManager) Start() error {
-	cm.Client.Handler = func(e Event) {
+func (sm *StreamManager) Start() error {
+	if sm.client == nil {
+		return errors.New("missing stream client")
+	}
+
+	handler := func(e Event) {
 		switch e.State {
 		case StateConnected:
-			cm.Metrics.setConnectTime()
+			sm.Metrics.setConnectTime()
 		case StateSessionEstablished:
-			cm.Metrics.setLoginTime()
+			sm.Metrics.setLoginTime()
 		case StateDisconnected:
 			// Reconnect on disconnection
-			cm.connect()
+			sm.connect()
 		case StateStreamError:
-			cm.Client.Disconnect()
+			sm.client.Disconnect()
 			// Only try reconnecting if we have not been kicked by another session to avoid connection loop.
 			if e.StreamError != "conflict" {
-				cm.connect()
+				sm.connect()
 			}
 		}
 	}
+	sm.client.SetHandler(handler)
 
-	return cm.connect()
+	return sm.connect()
 }
 
 // Stop cancels pending operations and terminates existing XMPP client.
-func (cm *ClientManager) Stop() {
+func (sm *StreamManager) Stop() {
 	// Remove on disconnect handler to avoid triggering reconnect
-	cm.Client.Handler = nil
-	cm.Client.Disconnect()
+	sm.client.SetHandler(nil)
+	sm.client.Disconnect()
 }
 
 // connect manages the reconnection loop and apply the define backoff to avoid overloading the server.
-func (cm *ClientManager) connect() error {
+func (sm *StreamManager) connect() error {
 	var backoff Backoff // TODO: Group backoff calculation features with connection manager?
 
 	for {
 		var err error
 		// TODO: Make it possible to define logger to log disconnect and reconnection attempts
-		cm.Metrics = initMetrics()
+		sm.Metrics = initMetrics()
 
-		if cm.Client.Session, err = cm.Client.Connect(); err != nil {
+		if err = sm.client.Connect(); err != nil {
 			var actualErr ConnError
 			if xerrors.As(err, &actualErr) {
 				if actualErr.Permanent {
@@ -81,13 +104,13 @@ func (cm *ClientManager) connect() error {
 		}
 	}
 
-	if cm.PostConnect != nil {
-		cm.PostConnect(cm.Client)
+	if sm.PostConnect != nil {
+		sm.PostConnect(sm.client)
 	}
 	return nil
 }
 
-// Client Metrics
+// Stream Metrics
 // ============================================================================
 
 type Metrics struct {


Test patch:
diff --git a/client_test.go b/client_test.go
index 85b9868..b7fb1ac 100644
--- a/client_test.go
+++ b/client_test.go
@@ -31,7 +31,7 @@ func TestClient_Connect(t *testing.T) {
 		t.Errorf("connect create XMPP client: %s", err)
 	}
 
-	if _, err = client.Connect(); err != nil {
+	if err = client.Connect(); err != nil {
 		t.Errorf("XMPP connection failed: %s", err)
 	}
 
@@ -52,7 +52,7 @@ func TestClient_NoInsecure(t *testing.T) {
 		t.Errorf("cannot create XMPP client: %s", err)
 	}
 
-	if _, err = client.Connect(); err == nil {
+	if err = client.Connect(); err == nil {
 		// When insecure is not allowed:
 		t.Errorf("should fail as insecure connection is not allowed and server does not support TLS")
 	}
diff --git a/component_test.go b/component_test.go
index 690d962..20e3772 100644
--- a/component_test.go
+++ b/component_test.go
@@ -3,10 +3,11 @@ package xmpp // import "gosrc.io/xmpp"
 import "testing"
 
 func TestHandshake(t *testing.T) {
-	c := Component{
-		Host:   "test.localhost",
+	opts := ComponentOptions{
+		Domain: "test.localhost",
 		Secret: "mypass",
 	}
+	c := Component{ComponentOptions: opts}
 
 	streamID := "1263952298440005243"
 	expected := "c77e2ef0109fbbc5161e83b51629cd1353495332"