Repository: ovn-org/libovsdb
Pull request #173 (2021-06-18 16:23:45)

client: Add Connected() and DisconnectNotify()
This allows a caller to see if we are currently connected.
It also allows a caller to receive a notification when the connection
has been closed so they may try to reconnect before getting
ErrNotConnected from an operation that needs to talk to the server

Signed-off-by: Dave Tucker <dave@dtucker.co.uk>


Patch:
diff --git a/client/client.go b/client/client.go
index d937c77..7aa406e 100644
--- a/client/client.go
+++ b/client/client.go
@@ -41,6 +41,8 @@ type Client interface {
 	Schema() *ovsdb.DatabaseSchema
 	Cache() *cache.TableCache
 	SetOption(Option) error
+	Connected() bool
+	DisconnectNotify() chan struct{}
 	Echo() error
 	Transact(...ovsdb.Operation) ([]ovsdb.OperationResult, error)
 	Monitor(jsonContext interface{}, t ...TableMonitor) error
@@ -52,15 +54,16 @@ type Client interface {
 
 // ovsdbClient is an OVSDB client
 type ovsdbClient struct {
-	options   *options
-	rpcClient *rpc2.Client
-	dbModel   *model.DBModel
-	schema    *ovsdb.DatabaseSchema
-	cache     *cache.TableCache
-	stopCh    chan struct{}
-	connected bool
-	api       API
-	mutex     sync.Mutex
+	options    *options
+	rpcClient  *rpc2.Client
+	dbModel    *model.DBModel
+	schema     *ovsdb.DatabaseSchema
+	cache      *cache.TableCache
+	stopCh     chan struct{}
+	connected  bool
+	disconnect chan struct{}
+	api        API
+	mutex      sync.Mutex
 }
 
 // NewOVSDBClient creates a new OVSDB Client with the provided
@@ -74,7 +77,8 @@ func NewOVSDBClient(databaseModel *model.DBModel, opts ...Option) (Client, error
 // newOVSDBClient creates a new ovsdbClient
 func newOVSDBClient(databaseModel *model.DBModel, opts ...Option) (*ovsdbClient, error) {
 	ovs := &ovsdbClient{
-		dbModel: databaseModel,
+		dbModel:    databaseModel,
+		disconnect: make(chan struct{}),
 	}
 	var err error
 	ovs.options, err = newOptions(opts...)
@@ -89,11 +93,11 @@ func newOVSDBClient(databaseModel *model.DBModel, opts ...Option) (*ovsdbClient,
 // The connection can be configured using one or more Option(s), like WithTLSConfig
 // If no WithEndpoint option is supplied, the default of unix:/var/run/openvswitch/ovsdb.sock is used
 func (o *ovsdbClient) Connect(ctx context.Context) error {
-	o.mutex.Lock()
-	defer o.mutex.Unlock()
 	if o.connected {
 		return nil
 	}
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
 	var c net.Conn
 	var dialer net.Dialer
 	var err error
@@ -218,6 +222,17 @@ func (o *ovsdbClient) SetOption(opt Option) error {
 	return opt(o.options)
 }
 
+// Connected returns whether or not the client is currently connected to the server
+func (o *ovsdbClient) Connected() bool {
+	return o.connected
+}
+
+// DisconnectNotify returns a channel which will notify the caller when the
+// server has disconnected
+func (o *ovsdbClient) DisconnectNotify() chan struct{} {
+	return o.disconnect
+}
+
 // RFC 7047 : Section 4.1.6 : Echo
 func (o *ovsdbClient) echo(args []interface{}, reply *[]interface{}) error {
 	*reply = args
@@ -410,6 +425,12 @@ func (o *ovsdbClient) handleDisconnectNotification() {
 	close(o.stopCh)
 	o.rpcClient = nil
 	o.cache = nil
+	select {
+	case o.disconnect <- struct{}{}:
+		// sent disconnect notification to client
+	default:
+		// client is not listening to the channel
+	}
 }
 
 // Disconnect will close the connection to the OVSDB server


Test patch:
diff --git a/client/ovs_integration_test.go b/client/ovs_integration_test.go
index db6dcff..1908746 100644
--- a/client/ovs_integration_test.go
+++ b/client/ovs_integration_test.go
@@ -111,11 +111,39 @@ func TestConnectReconnectIntegration(t *testing.T) {
 	err = ovs.Connect(ctx)
 	require.NoError(t, err)
 
+	assert.Equal(t, true, ovs.Connected())
+
 	err = ovs.Echo()
 	require.NoError(t, err)
 
+	// make another connect call, this should return without error as we're already connected
+	ctx, cancel = context.WithTimeout(context.Background(), 2*time.Second)
+	defer cancel()
+	err = ovs.Connect(ctx)
+	require.NoError(t, err)
+
+	disconnectNotification := ovs.DisconnectNotify()
+	notified := make(chan struct{})
+	ready := make(chan struct{})
+
+	go func() {
+		ready <- struct{}{}
+		<-disconnectNotification
+		notified <- struct{}{}
+	}()
+
+	<-ready
 	ovs.Disconnect()
 
+	select {
+	case <-notified:
+		// got notification
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected a disconnect notification but didn't receive one")
+	}
+
+	assert.Equal(t, false, ovs.Connected())
+
 	err = ovs.Echo()
 	require.EqualError(t, err, ErrNotConnected.Error())
 
@@ -124,6 +152,7 @@ func TestConnectReconnectIntegration(t *testing.T) {
 
 	err = ovs.Connect(ctx)
 	require.NoError(t, err)
+	defer ovs.Disconnect()
 
 	err = ovs.Echo()
 	assert.NoError(t, err)