Use StreamClient interface in StreamManager

This commit is contained in:
Mickael Remond 2019-06-08 18:52:19 +02:00 committed by Mickaël Rémond
parent 021f6d3740
commit 736a60cd1b
5 changed files with 40 additions and 31 deletions

View File

@ -2,4 +2,10 @@ module gosrc.io/xmpp/_examples
go 1.12 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

View File

@ -1,11 +1,8 @@
github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= 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/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= 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/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 h1:bhOzK9QyoD0ogCnFro1m2mz41+Ib0oOhfJnBp5MR4K4=
golang.org/x/xerrors v0.0.0-20190513163551-3ee3066db522/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= 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-20190608160922-63a29d5c218a h1:TKUhPFlQkBLpoSTNgh4rX64T1FmObGcggIYl7d2q4vM=
gosrc.io/xmpp v0.0.0-20190608091551-b7461ae97fed/go.mod h1:it3z4S42Sy7eHWFqwmdFJbygg7lCmbrhKeqK7HQSqSU= gosrc.io/xmpp v0.0.0-20190608160922-63a29d5c218a/go.mod h1:6NJG4vRCxQJMGLxIdroPLPd++FPLOmDqJdJEt2mu4kQ=

View File

@ -143,7 +143,7 @@ func (c *Client) Connect() error {
} }
c.updateState(StateConnected) 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 { if c.conn, c.Session, err = NewSession(c.conn, c.config); err != nil {
return err return err
} }

View File

@ -12,7 +12,7 @@ type Config struct {
Password string Password string
PacketLogger *os.File // Used for debugging PacketLogger *os.File // Used for debugging
Lang string // TODO: should default to 'en' 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 // 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. // is supported on the server, we will still try to use it.
Insecure bool Insecure bool

View File

@ -1,6 +1,7 @@
package xmpp // import "gosrc.io/xmpp" package xmpp // import "gosrc.io/xmpp"
import ( import (
"errors"
"time" "time"
"golang.org/x/xerrors" "golang.org/x/xerrors"
@ -11,13 +12,13 @@ import (
// stream events and doing the right operations. // stream events and doing the right operations.
// //
// It can handle: // It can handle:
// - Connection // - Client
// - Stream establishment workflow // - Stream establishment workflow
// - Reconnection strategies, with exponential backoff. It also takes into account // - Reconnection strategies, with exponential backoff. It also takes into account
// permanent errors to avoid useless reconnection loops. // permanent errors to avoid useless reconnection loops.
// - Metrics processing // - Metrics processing
type StreamSession interface { type StreamClient interface {
Connect() error Connect() error
Disconnect() Disconnect()
SetHandler(handler EventHandler) SetHandler(handler EventHandler)
@ -26,66 +27,71 @@ type StreamSession interface {
// StreamManager 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. // apply reconnection strategy.
type StreamManager struct { type StreamManager struct {
Client *Client client StreamClient
Session *Session
PostConnect PostConnect PostConnect PostConnect
// Store low level metrics // Store low level metrics
Metrics *Metrics Metrics *Metrics
} }
type PostConnect func(c *Client) type PostConnect func(c StreamClient)
// NewStreamManager creates a new StreamManager structure, intended to support // NewStreamManager creates a new StreamManager structure, intended to support
// handling XMPP client state event changes and auto-trigger reconnection // handling XMPP client state event changes and auto-trigger reconnection
// based on StreamManager configuration. // based on StreamManager configuration.
func NewStreamManager(client *Client, pc PostConnect) *StreamManager { // TODO: Move parameters to Start and remove factory method
func NewStreamManager(client StreamClient, pc PostConnect) *StreamManager {
return &StreamManager{ return &StreamManager{
Client: client, client: client,
PostConnect: pc, PostConnect: pc,
} }
} }
// Start launch the connection loop // Start launch the connection loop
func (cm *StreamManager) Start() error { func (sm *StreamManager) Start() error {
cm.Client.Handler = func(e Event) { if sm.client == nil {
switch e.State { return errors.New("missing stream client")
case StateConnected:
cm.Metrics.setConnectTime()
case StateSessionEstablished:
cm.Metrics.setLoginTime()
case StateDisconnected:
// Reconnect on disconnection
cm.connect()
case StateStreamError:
cm.Client.Disconnect()
// Only try reconnecting if we have not been kicked by another session to avoid connection loop.
if e.StreamError != "conflict" {
cm.connect()
}
}
} }
return cm.connect() handler := func(e Event) {
switch e.State {
case StateConnected:
sm.Metrics.setConnectTime()
case StateSessionEstablished:
sm.Metrics.setLoginTime()
case StateDisconnected:
// Reconnect on disconnection
sm.connect()
case StateStreamError:
sm.client.Disconnect()
// Only try reconnecting if we have not been kicked by another session to avoid connection loop.
if e.StreamError != "conflict" {
sm.connect()
}
}
}
sm.client.SetHandler(handler)
return sm.connect()
} }
// Stop cancels pending operations and terminates existing XMPP client. // Stop cancels pending operations and terminates existing XMPP client.
func (cm *StreamManager) Stop() { func (sm *StreamManager) Stop() {
// Remove on disconnect handler to avoid triggering reconnect // Remove on disconnect handler to avoid triggering reconnect
cm.Client.Handler = nil sm.client.SetHandler(nil)
cm.Client.Disconnect() sm.client.Disconnect()
} }
// connect manages the reconnection loop and apply the define backoff to avoid overloading the server. // connect manages the reconnection loop and apply the define backoff to avoid overloading the server.
func (cm *StreamManager) connect() error { func (sm *StreamManager) connect() error {
var backoff Backoff // TODO: Group backoff calculation features with connection manager? var backoff Backoff // TODO: Group backoff calculation features with connection manager?
for { for {
var err error var err error
// TODO: Make it possible to define logger to log disconnect and reconnection attempts // TODO: Make it possible to define logger to log disconnect and reconnection attempts
cm.Metrics = initMetrics() sm.Metrics = initMetrics()
if err = cm.Client.Connect(); err != nil { if err = sm.client.Connect(); err != nil {
var actualErr ConnError var actualErr ConnError
if xerrors.As(err, &actualErr) { if xerrors.As(err, &actualErr) {
if actualErr.Permanent { if actualErr.Permanent {
@ -98,8 +104,8 @@ func (cm *StreamManager) connect() error {
} }
} }
if cm.PostConnect != nil { if sm.PostConnect != nil {
cm.PostConnect(cm.Client) sm.PostConnect(sm.client)
} }
return nil return nil
} }