Repository: ovn-org/libovsdb (go)

Issue:
client: Support Multiple Monitor Requests (and maybe resuming Monitor on reconnect)
After #166 has merged we should consider supporting multiple monitors and potentially resuming them on reconnect.
If we could resume on reconnect, then we could quite possibly add an option `WithReconnect` to reduce the amount of user code to deal with handling disconnects. We'd need to consider whether there would be a Deadline for reconnect before we return an error back to the client, and what that might look like `WithReconnet(timeout time.Duration, err chan error)`?

Implementation-wise we'd need to think about the following:

1. We should enforce that the `jsonContext` for each Monitor is unique
1. Based on :point_up: we might actually want to generate one and return it to the client. E.g `func (o *ovsdbClient) Monitor(...) (uuid, error)` so it can be used for `MonitorCancel`
1. We should store all current `Monitor` with options in a map on the `ovsdbClient` - this will be needed for `update()` to know which monitor request an update is for. It may also be useful to re-establish monitors after the client has been disconnected

Pull request:
Auto reconnect
This commit builds on the work released in v0.5.0 to provide disconnect notifications.
It adds the `WithReconnect` option that provides the ability for the library to handle automatically reconnecting then client when it is disconnected. This includes re-establishing any Monitors that a user may have established.

**Normal Operation**
- `Disconnect()` and `Close()` API calls do the same thing - clear all connection state and set the cache to nil
- In both cases, you'll get a notification on the `DisconnectNotify` channel

**WithReconnect**
- `Disconnect()` (or anything that closes the underlying `rpcClient`) will trigger reconnection. You will not get a notification on the `DisconnectNotify` channel. The cache content is cleared BUT any event handlers will be retained.
- `Close()` will do a full shutdown of the client. All connection state is cleared and the cache is set to nil. You will get a notification on the `DisocnnectNotify` channel

`WithReconnect` includes a timeout (for the call to `DialWithContext` and a user-provided backoff function, which can handle, `ZeroBackoff`, `ExponentialBackoff` and more.

**API Changes**

- `Monitor` and `MonitorAll` do not accept a `jsonContext` argument any more. We instead generate a unique ID for every monitor and return this to the caller. This can then be used to `MonitorCancel` if they wish. This is required to enable the auto-reconnect feature because we want to ensure the unique-ness of monitor id's across connection hiccups.
- `Cache` gains a `Purge` method to drop the contents of the cache but leave any eventHandlers intact so we can ensure we don't have stale entries in the cache on re-establishing a monitor.

** Misc **
This also cleans adds locking of key state in the `client` ensure that things don't break catastrophically when the automatic reconnection is happening in a separate goroutine. This should also make the client safe to be shared across by more than one goroutine.

Fixes #172 

Patch:
diff --git a/cache/cache.go b/cache/cache.go
index 443eddb..bcc2204 100644
--- a/cache/cache.go
+++ b/cache/cache.go
@@ -82,10 +82,6 @@ type RowCache struct {
 func (r *RowCache) Row(uuid string) model.Model {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
-	return r.row(uuid)
-}
-
-func (r *RowCache) row(uuid string) model.Model {
 	if row, ok := r.cache[uuid]; ok {
 		return row.(model.Model)
 	}
@@ -96,10 +92,6 @@ func (r *RowCache) row(uuid string) model.Model {
 func (r *RowCache) Create(uuid string, m model.Model) error {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
-	return r.create(uuid, m)
-}
-
-func (r *RowCache) create(uuid string, m model.Model) error {
 	if _, ok := r.cache[uuid]; ok {
 		return fmt.Errorf("row %s already exists", uuid)
 	}
@@ -138,10 +130,6 @@ func (r *RowCache) create(uuid string, m model.Model) error {
 func (r *RowCache) Update(uuid string, m model.Model) error {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
-	return r.update(uuid, m)
-}
-
-func (r *RowCache) update(uuid string, m model.Model) error {
 	if _, ok := r.cache[uuid]; !ok {
 		return fmt.Errorf("row %s does not exist", uuid)
 	}
@@ -210,10 +198,6 @@ func (r *RowCache) update(uuid string, m model.Model) error {
 func (r *RowCache) Delete(uuid string) error {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
-	return r.delete(uuid)
-}
-
-func (r *RowCache) delete(uuid string) error {
 	if _, ok := r.cache[uuid]; !ok {
 		return fmt.Errorf("row %s does not exist", uuid)
 	}
@@ -246,8 +230,8 @@ func (r *RowCache) Rows() []string {
 
 // Len returns the length of the cache
 func (r *RowCache) Len() int {
-	r.mutex.Lock()
-	defer r.mutex.Unlock()
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
 	return len(r.cache)
 }
 
@@ -306,7 +290,9 @@ type TableCache struct {
 	eventProcessor *eventProcessor
 	mapper         *mapper.Mapper
 	dbModel        *model.DBModel
+	schema         *ovsdb.DatabaseSchema
 	ovsdb.NotificationHandler
+	mutex sync.RWMutex
 }
 
 // Data is the type for data that can be prepoulated in the cache
@@ -335,9 +321,11 @@ func NewTableCache(schema *ovsdb.DatabaseSchema, dbModel *model.DBModel, data Da
 	}
 	return &TableCache{
 		cache:          cache,
+		schema:         schema,
 		eventProcessor: eventProcessor,
 		mapper:         mapper.NewMapper(schema),
 		dbModel:        dbModel,
+		mutex:          sync.RWMutex{},
 	}, nil
 }
 
@@ -353,6 +341,8 @@ func (t *TableCache) DBModel() *model.DBModel {
 
 // Table returns the a Table from the cache with a given name
 func (t *TableCache) Table(name string) *RowCache {
+	t.mutex.RLock()
+	defer t.mutex.RUnlock()
 	if table, ok := t.cache[name]; ok {
 		return table
 	}
@@ -361,6 +351,8 @@ func (t *TableCache) Table(name string) *RowCache {
 
 // Tables returns a list of table names that are in the cache
 func (t *TableCache) Tables() []string {
+	t.mutex.RLock()
+	defer t.mutex.RUnlock()
 	var result []string
 	for k := range t.cache {
 		result = append(result, k)
@@ -393,24 +385,10 @@ func (t *TableCache) Echo([]interface{}) {
 func (t *TableCache) Disconnected() {
 }
 
-// lock acquires a lock on all tables in the cache
-func (t *TableCache) lock() {
-	for _, r := range t.cache {
-		r.mutex.Lock()
-	}
-}
-
-// unlock releases a lock on all tables in the cache
-func (t *TableCache) unlock() {
-	for _, r := range t.cache {
-		r.mutex.Unlock()
-	}
-}
-
 // Populate adds data to the cache and places an event on the channel
 func (t *TableCache) Populate(tableUpdates ovsdb.TableUpdates) {
-	t.lock()
-	defer t.unlock()
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
 	for table := range t.dbModel.Types() {
 		updates, ok := tableUpdates[table]
 		if !ok {
@@ -423,9 +401,9 @@ func (t *TableCache) Populate(tableUpdates ovsdb.TableUpdates) {
 				if err != nil {
 					panic(err)
 				}
-				if existing := tCache.row(uuid); existing != nil {
+				if existing := tCache.Row(uuid); existing != nil {
 					if !reflect.DeepEqual(newModel, existing) {
-						if err := tCache.update(uuid, newModel); err != nil {
+						if err := tCache.Update(uuid, newModel); err != nil {
 							panic(err)
 						}
 						t.eventProcessor.AddEvent(updateEvent, table, existing, newModel)
@@ -433,7 +411,7 @@ func (t *TableCache) Populate(tableUpdates ovsdb.TableUpdates) {
 					// no diff
 					continue
 				}
-				if err := tCache.create(uuid, newModel); err != nil {
+				if err := tCache.Create(uuid, newModel); err != nil {
 					panic(err)
 				}
 				t.eventProcessor.AddEvent(addEvent, table, nil, newModel)
@@ -443,7 +421,7 @@ func (t *TableCache) Populate(tableUpdates ovsdb.TableUpdates) {
 				if err != nil {
 					panic(err)
 				}
-				if err := tCache.delete(uuid); err != nil {
+				if err := tCache.Delete(uuid); err != nil {
 					panic(err)
 				}
 				t.eventProcessor.AddEvent(deleteEvent, table, oldModel, nil)
@@ -453,6 +431,17 @@ func (t *TableCache) Populate(tableUpdates ovsdb.TableUpdates) {
 	}
 }
 
+// Purge drops all data in the cache and reinitializes it using the
+// provided schema
+func (t *TableCache) Purge(schema *ovsdb.DatabaseSchema) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	tableTypes := t.dbModel.Types()
+	for name, tableSchema := range t.schema.Tables {
+		t.cache[name] = newRowCache(name, tableSchema, tableTypes[name])
+	}
+}
+
 // AddEventHandler registers the supplied EventHandler to receive cache events
 func (t *TableCache) AddEventHandler(handler EventHandler) {
 	t.eventProcessor.AddEventHandler(handler)
diff --git a/client/client.go b/client/client.go
index 660f411..5a95e54 100644
--- a/client/client.go
+++ b/client/client.go
@@ -12,8 +12,10 @@ import (
 	"strings"
 	"sync"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/cenkalti/rpc2"
 	"github.com/cenkalti/rpc2/jsonrpc"
+	"github.com/google/uuid"
 	"github.com/ovn-org/libovsdb/cache"
 	"github.com/ovn-org/libovsdb/mapper"
 	"github.com/ovn-org/libovsdb/model"
@@ -38,6 +40,7 @@ var ErrNotConnected = errors.New("not connected")
 type Client interface {
 	Connect(context.Context) error
 	Disconnect()
+	Close()
 	Schema() *ovsdb.DatabaseSchema
 	Cache() *cache.TableCache
 	SetOption(Option) error
@@ -45,25 +48,30 @@ type Client interface {
 	DisconnectNotify() chan struct{}
 	Echo() error
 	Transact(...ovsdb.Operation) ([]ovsdb.OperationResult, error)
-	Monitor(jsonContext interface{}, t ...TableMonitor) error
-	MonitorAll(jsonContext interface{}) error
-	MonitorCancel(jsonContext interface{}) error
+	Monitor(...TableMonitor) (string, error)
+	MonitorAll() (string, error)
+	MonitorCancel(id string) error
 	NewTableMonitor(m model.Model, fields ...interface{}) TableMonitor
 	API
 }
 
 // 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{}
-	disconnect chan struct{}
-	api        API
-	// mutex protects the rpcClient from concurrent access
-	mutex sync.Mutex
+	options       *options
+	rpcClient     *rpc2.Client
+	rpcMutex      sync.RWMutex
+	dbModel       *model.DBModel
+	schema        *ovsdb.DatabaseSchema
+	schemaMutex   sync.RWMutex
+	cache         *cache.TableCache
+	cacheMutex    sync.RWMutex
+	stopCh        chan struct{}
+	disconnect    chan struct{}
+	api           API
+	monitors      map[string][]TableMonitor
+	monitorsMutex sync.Mutex
+	shutdown      bool
+	shutdownMutex sync.Mutex
 }
 
 // NewOVSDBClient creates a new OVSDB Client with the provided
@@ -77,8 +85,12 @@ 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,
-		disconnect: make(chan struct{}),
+		dbModel:       databaseModel,
+		disconnect:    make(chan struct{}),
+		rpcMutex:      sync.RWMutex{},
+		schemaMutex:   sync.RWMutex{},
+		monitorsMutex: sync.Mutex{},
+		shutdownMutex: sync.Mutex{},
 	}
 	var err error
 	ovs.options, err = newOptions(opts...)
@@ -93,8 +105,12 @@ 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()
+	return o.connect(ctx, false)
+}
+
+func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
+	o.rpcMutex.Lock()
+	defer o.rpcMutex.Unlock()
 	if o.rpcClient != nil {
 		return nil
 	}
@@ -162,8 +178,15 @@ func (o *ovsdbClient) Connect(ctx context.Context) error {
 			strings.Join(combined, ". "))
 	}
 
-	if err == nil {
-		o.schema = schema
+	if err != nil {
+		o.rpcClient.Close()
+		return err
+	}
+	o.schemaMutex.Lock()
+	o.schema = schema
+	o.schemaMutex.Unlock()
+	if o.cache == nil {
+		o.cacheMutex.Lock()
 		if cache, err := cache.NewTableCache(schema, o.dbModel, nil); err == nil {
 			o.cache = cache
 			o.api = newAPI(o.cache)
@@ -171,17 +194,37 @@ func (o *ovsdbClient) Connect(ctx context.Context) error {
 			o.rpcClient.Close()
 			return err
 		}
+		o.cacheMutex.Unlock()
 	} else {
-		o.rpcClient.Close()
-		return err
+		// purge cache contents and ensure we are using latest schema
+		// cache event handlers are untouched
+		o.cache.Purge(schema)
+	}
+
+	if reconnect {
+		o.monitorsMutex.Lock()
+		defer o.monitorsMutex.Unlock()
+		for id, request := range o.monitors {
+			err = o.monitor(id, reconnect, request...)
+			if err != nil {
+				o.rpcClient.Close()
+				return err
+			}
+		}
+	} else {
+		o.monitorsMutex.Lock()
+		defer o.monitorsMutex.Unlock()
+		o.monitors = make(map[string][]TableMonitor)
 	}
+
+	go o.handleDisconnectNotification()
 	go o.cache.Run(o.stopCh)
 	return nil
 }
 
 // createRPC2Client creates an rpcClient using the provided connection
 // It is also responsible for setting up go routines for client-side event handling
-// and handling disconnects. Should only be called when the mutex is held
+// Should only be called when the mutex is held
 func (o *ovsdbClient) createRPC2Client(conn net.Conn) error {
 	o.stopCh = make(chan struct{})
 	o.rpcClient = rpc2.NewClientWithCodec(jsonrpc.NewJSONCodec(conn))
@@ -193,13 +236,14 @@ func (o *ovsdbClient) createRPC2Client(conn net.Conn) error {
 		return o.update(args, reply)
 	})
 	go o.rpcClient.Run()
-	go o.handleDisconnectNotification()
 	return nil
 }
 
 // Schema returns the DatabaseSchema that is being used by the client
 // it will be nil until a connection has been established
 func (o *ovsdbClient) Schema() *ovsdb.DatabaseSchema {
+	o.schemaMutex.RLock()
+	defer o.schemaMutex.RUnlock()
 	return o.schema
 }
 
@@ -207,14 +251,16 @@ func (o *ovsdbClient) Schema() *ovsdb.DatabaseSchema {
 // ovsdb update notifications. It will be nil until a connection
 // has been established, and empty unless you call Monitor
 func (o *ovsdbClient) Cache() *cache.TableCache {
+	o.cacheMutex.RLock()
+	defer o.cacheMutex.RUnlock()
 	return o.cache
 }
 
 // SetOption sets a new value for an option.
 // It may only be called when the client is not connected
 func (o *ovsdbClient) SetOption(opt Option) error {
-	o.mutex.Lock()
-	defer o.mutex.Unlock()
+	o.rpcMutex.RLock()
+	defer o.rpcMutex.RUnlock()
 	if o.rpcClient != nil {
 		return fmt.Errorf("cannot set option when client is connected")
 	}
@@ -223,8 +269,8 @@ func (o *ovsdbClient) SetOption(opt Option) error {
 
 // Connected returns whether or not the client is currently connected to the server
 func (o *ovsdbClient) Connected() bool {
-	o.mutex.Lock()
-	defer o.mutex.Unlock()
+	o.rpcMutex.RLock()
+	defer o.rpcMutex.RUnlock()
 	return o.rpcClient != nil
 }
 
@@ -256,7 +302,9 @@ func (o *ovsdbClient) update(args []json.RawMessage, reply *[]interface{}) error
 		return err
 	}
 	// Update the local DB cache with the tableUpdates
+	o.cacheMutex.RLock()
 	o.cache.Update(value, updates)
+	o.cacheMutex.RUnlock()
 	*reply = []interface{}{}
 	return nil
 }
@@ -269,6 +317,9 @@ func (o *ovsdbClient) getSchema(dbName string) (*ovsdb.DatabaseSchema, error) {
 	var reply ovsdb.DatabaseSchema
 	err := o.rpcClient.Call("get_schema", args, &reply)
 	if err != nil {
+		if err == rpc2.ErrShutdown {
+			return nil, ErrNotConnected
+		}
 		return nil, err
 	}
 	return &reply, err
@@ -281,6 +332,9 @@ func (o *ovsdbClient) listDbs() ([]string, error) {
 	var dbs []string
 	err := o.rpcClient.Call("list_dbs", nil, &dbs)
 	if err != nil {
+		if err == rpc2.ErrShutdown {
+			return nil, ErrNotConnected
+		}
 		return nil, fmt.Errorf("listdbs failure - %v", err)
 	}
 	return dbs, err
@@ -290,51 +344,59 @@ func (o *ovsdbClient) listDbs() ([]string, error) {
 // RFC 7047 : transact
 func (o *ovsdbClient) Transact(operation ...ovsdb.Operation) ([]ovsdb.OperationResult, error) {
 	var reply []ovsdb.OperationResult
-
-	if ok := o.schema.ValidateOperations(operation...); !ok {
+	if ok := o.Schema().ValidateOperations(operation...); !ok {
 		return nil, fmt.Errorf("validation failed for the operation")
 	}
-
 	args := ovsdb.NewTransactArgs(o.schema.Name, operation...)
-	o.mutex.Lock()
+
+	o.rpcMutex.Lock()
 	if o.rpcClient == nil {
-		o.mutex.Unlock()
+		o.rpcMutex.Unlock()
 		return nil, ErrNotConnected
 	}
 	err := o.rpcClient.Call("transact", args, &reply)
-	o.mutex.Unlock()
+	o.rpcMutex.Unlock()
 	if err != nil {
+		if err == rpc2.ErrShutdown {
+			return nil, ErrNotConnected
+		}
 		return nil, err
 	}
 	return reply, nil
 }
 
 // MonitorAll is a convenience method to monitor every table/column
-func (o *ovsdbClient) MonitorAll(jsonContext interface{}) error {
+func (o *ovsdbClient) MonitorAll() (string, error) {
 	var options []TableMonitor
 	for name := range o.dbModel.Types() {
 		options = append(options, TableMonitor{Table: name})
 	}
-	return o.Monitor(jsonContext, options...)
+	return o.Monitor(options...)
 }
 
 // MonitorCancel will request cancel a previously issued monitor request
 // RFC 7047 : monitor_cancel
-func (o *ovsdbClient) MonitorCancel(jsonContext interface{}) error {
+func (o *ovsdbClient) MonitorCancel(id string) error {
 	var reply ovsdb.OperationResult
-	args := ovsdb.NewMonitorCancelArgs(jsonContext)
-	o.mutex.Lock()
-	defer o.mutex.Unlock()
+	args := ovsdb.NewMonitorCancelArgs(id)
+	o.rpcMutex.Lock()
+	defer o.rpcMutex.Unlock()
 	if o.rpcClient == nil {
 		return ErrNotConnected
 	}
 	err := o.rpcClient.Call("monitor_cancel", args, &reply)
 	if err != nil {
+		if err == rpc2.ErrShutdown {
+			return ErrNotConnected
+		}
 		return err
 	}
 	if reply.Error != "" {
 		return fmt.Errorf("error while executing transaction: %s", reply.Error)
 	}
+	o.monitorsMutex.Lock()
+	defer o.monitorsMutex.Unlock()
+	delete(o.monitors, id)
 	return nil
 }
 
@@ -366,12 +428,17 @@ func (o *ovsdbClient) NewTableMonitor(m model.Model, fields ...interface{}) Tabl
 // and populate the cache with them. Subsequent updates will be processed
 // by the Update Notifications
 // RFC 7047 : monitor
-func (o *ovsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor) error {
+func (o *ovsdbClient) Monitor(options ...TableMonitor) (string, error) {
+	id := uuid.NewString()
+	return id, o.monitor(id, false, options...)
+}
+
+func (o *ovsdbClient) monitor(id string, reconnect bool, options ...TableMonitor) error {
 	if len(options) == 0 {
 		return fmt.Errorf("no monitor options provided")
 	}
 	var reply ovsdb.TableUpdates
-	mapper := mapper.NewMapper(o.schema)
+	mapper := mapper.NewMapper(o.Schema())
 	typeMap := o.dbModel.Types()
 	requests := make(map[string]ovsdb.MonitorRequest)
 	for _, o := range options {
@@ -388,16 +455,26 @@ func (o *ovsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor)
 		}
 		requests[o.Table] = *request
 	}
-	args := ovsdb.NewMonitorArgs(o.schema.Name, jsonContext, requests)
-	o.mutex.Lock()
-	defer o.mutex.Unlock()
+	args := ovsdb.NewMonitorArgs(o.Schema().Name, id, requests)
+	if !reconnect {
+		o.rpcMutex.RLock()
+		defer o.rpcMutex.RUnlock()
+	}
 	if o.rpcClient == nil {
 		return ErrNotConnected
 	}
 	err := o.rpcClient.Call("monitor", args, &reply)
 	if err != nil {
+		if err == rpc2.ErrShutdown {
+			return ErrNotConnected
+		}
 		return err
 	}
+	if !reconnect {
+		o.monitorsMutex.Lock()
+		defer o.monitorsMutex.Unlock()
+		o.monitors[id] = options
+	}
 	o.cache.Populate(reply)
 	return nil
 }
@@ -406,14 +483,16 @@ func (o *ovsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor)
 func (o *ovsdbClient) Echo() error {
 	args := ovsdb.NewEchoArgs()
 	var reply []interface{}
-	o.mutex.Lock()
-	defer o.mutex.Unlock()
+	o.rpcMutex.RLock()
+	defer o.rpcMutex.RUnlock()
 	if o.rpcClient == nil {
 		return ErrNotConnected
 	}
 	err := o.rpcClient.Call("echo", args, &reply)
 	if err != nil {
-		return err
+		if err == rpc2.ErrShutdown {
+			return ErrNotConnected
+		}
 	}
 	if !reflect.DeepEqual(args, reply) {
 		return fmt.Errorf("incorrect server response: %v, %v", args, reply)
@@ -423,11 +502,47 @@ func (o *ovsdbClient) Echo() error {
 
 func (o *ovsdbClient) handleDisconnectNotification() {
 	<-o.rpcClient.DisconnectNotify()
-	o.mutex.Lock()
-	defer o.mutex.Unlock()
-	o.rpcClient = nil
+	// close the stopCh, which will stop the cache event processor
 	close(o.stopCh)
+	o.rpcMutex.Lock()
+	if o.options.reconnect && !o.shutdown {
+		o.rpcClient = nil
+		o.rpcMutex.Unlock()
+		connect := func() error {
+			ctx, cancel := context.WithTimeout(context.Background(), o.options.timeout)
+			defer cancel()
+			return o.connect(ctx, true)
+		}
+		err := backoff.Retry(connect, o.options.backoff)
+		if err != nil {
+			// TODO: We should look at passing this back to the
+			// caller to handle
+			panic(err)
+		}
+		// this goroutine finishes, and is replaced with a new one (from Connect)
+		return
+	}
+
+	// clear connection state
+	o.rpcClient = nil
+	o.rpcMutex.Unlock()
+
+	o.cacheMutex.Lock()
+	defer o.cacheMutex.Unlock()
 	o.cache = nil
+
+	o.schemaMutex.Lock()
+	defer o.schemaMutex.Unlock()
+	o.schema = nil
+
+	o.monitorsMutex.Lock()
+	defer o.monitorsMutex.Unlock()
+	o.monitors = nil
+
+	o.shutdownMutex.Lock()
+	defer o.shutdownMutex.Unlock()
+	o.shutdown = false
+
 	select {
 	case o.disconnect <- struct{}{}:
 		// sent disconnect notification to client
@@ -437,12 +552,29 @@ func (o *ovsdbClient) handleDisconnectNotification() {
 }
 
 // Disconnect will close the connection to the OVSDB server
+// If the client was created with WithReconnect then the client
+// will reconnect afterwards
 func (o *ovsdbClient) Disconnect() {
-	o.mutex.Lock()
-	defer o.mutex.Unlock()
+	o.rpcMutex.Lock()
+	defer o.rpcMutex.Unlock()
+	if o.rpcClient == nil {
+		return
+	}
+	o.rpcClient.Close()
+}
+
+// Close will close the connection to the OVSDB server
+// It will remove all stored state ready for the next connection
+// Even If the client was created with WithReconnect it will not reconnect afterwards
+func (o *ovsdbClient) Close() {
+	o.rpcMutex.Lock()
+	defer o.rpcMutex.Unlock()
 	if o.rpcClient == nil {
 		return
 	}
+	o.shutdownMutex.Lock()
+	defer o.shutdownMutex.Unlock()
+	o.shutdown = true
 	o.rpcClient.Close()
 }
 
diff --git a/client/options.go b/client/options.go
index d73f95d..a57d341 100644
--- a/client/options.go
+++ b/client/options.go
@@ -3,6 +3,9 @@ package client
 import (
 	"crypto/tls"
 	"net/url"
+	"time"
+
+	"github.com/cenkalti/backoff/v4"
 )
 
 const (
@@ -14,6 +17,9 @@ const (
 type options struct {
 	endpoints []string
 	tlsConfig *tls.Config
+	reconnect bool
+	timeout   time.Duration
+	backoff   backoff.BackOff
 }
 
 type Option func(o *options) error
@@ -72,3 +78,16 @@ func WithEndpoint(endpoint string) Option {
 		return nil
 	}
 }
+
+// WithReconnect tells the client to automatically reconnect when
+// disconnected. The timeout is used to construct the context on
+// each call to Connect, while backoff dicates the backoff
+// algorithm to use
+func WithReconnect(timeout time.Duration, backoff backoff.BackOff) Option {
+	return func(o *options) error {
+		o.reconnect = true
+		o.timeout = timeout
+		o.backoff = backoff
+		return nil
+	}
+}
diff --git a/cmd/stress/stress.go b/cmd/stress/stress.go
index 50b89c0..e0094b7 100644
--- a/cmd/stress/stress.go
+++ b/cmd/stress/stress.go
@@ -64,7 +64,7 @@ func cleanup(ctx context.Context) {
 	}
 	defer ovs.Disconnect()
 
-	if err := ovs.MonitorAll(""); err != nil {
+	if _, err := ovs.MonitorAll(); err != nil {
 		log.Fatal(err)
 	}
 
@@ -136,7 +136,7 @@ func run(ctx context.Context, resultsChan chan result, wg *sync.WaitGroup) {
 		},
 	)
 
-	if err := ovs.MonitorAll(""); err != nil {
+	if _, err := ovs.MonitorAll(); err != nil {
 		log.Fatal(err)
 	}
 
diff --git a/example/play_with_ovs/play_with_ovs.go b/example/play_with_ovs/play_with_ovs.go
index 7861394..384d7ab 100644
--- a/example/play_with_ovs/play_with_ovs.go
+++ b/example/play_with_ovs/play_with_ovs.go
@@ -119,7 +119,7 @@ func main() {
 			}
 		},
 	})
-	err = ovs.Monitor("play_with_ovs",
+	_, err = ovs.Monitor(
 		ovs.NewTableMonitor(&OpenvSwitch{}),
 		ovs.NewTableMonitor(&Bridge{}),
 	)
diff --git a/go.mod b/go.mod
index 85ab368..66bd069 100644
--- a/go.mod
+++ b/go.mod
@@ -4,6 +4,7 @@ go 1.16
 
 require (
 	github.com/cenk/hub v1.0.1 // indirect
+	github.com/cenkalti/backoff/v4 v4.1.1
 	github.com/cenkalti/hub v1.0.1 // indirect
 	github.com/cenkalti/rpc2 v0.0.0-20210220005819-4a29bc83afe1
 	github.com/google/uuid v1.2.0
diff --git a/go.sum b/go.sum
index 231f21d..4ad68f7 100644
--- a/go.sum
+++ b/go.sum
@@ -6,8 +6,9 @@ github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5 h1:TngWCqHvy9oXAN6lEV
 github.com/Nvveen/Gotty v0.0.0-20120604004816-cd527374f1e5/go.mod h1:lmUJ/7eu/Q8D7ML55dXQrVaamCz2vxCfdQBasLZfHKk=
 github.com/cenk/hub v1.0.1 h1:RBwXNOF4a8KjD8BJ08XqN8KbrqaGiQLDrgvUGJSHuPA=
 github.com/cenk/hub v1.0.1/go.mod h1:rJM1LNAW0ppT8FMMuPK6c2NP/R2nH/UthtuRySSaf6Y=
-github.com/cenkalti/backoff/v4 v4.1.0 h1:c8LkOFQTzuO0WBM/ae5HdGQuZPfPxp7lqBRwQRm4fSc=
 github.com/cenkalti/backoff/v4 v4.1.0/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
+github.com/cenkalti/backoff/v4 v4.1.1 h1:G2HAfAmvm/GcKan2oOQpBXOd2tT2G57ZnZGWa1PxPBQ=
+github.com/cenkalti/backoff/v4 v4.1.1/go.mod h1:scbssz8iZGpm3xbr14ovlUdkxfGXNInqkPWOWmG2CLw=
 github.com/cenkalti/hub v1.0.1 h1:UMtjc6dHSaOQTO15SVA50MBIR9zQwvsukQupDrkIRtg=
 github.com/cenkalti/hub v1.0.1/go.mod h1:tcYwtS3a2d9NO/0xDXVJWx3IedurUjYCqFCmpi0lpHs=
 github.com/cenkalti/rpc2 v0.0.0-20210220005819-4a29bc83afe1 h1:aT9Ez2drLmrviqTnVnH87AeXLXLgUrXACJ2g90cTT2w=


Test patch:
diff --git a/client/client_test.go b/client/client_test.go
index 23967cd..ebd93fe 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -758,7 +758,8 @@ func TestOperationWhenNotConnected(t *testing.T) {
 		{
 			"monitor/monitor all",
 			func() error {
-				return ovs.MonitorAll("")
+				_, err := ovs.MonitorAll()
+				return err
 			},
 		},
 		{
diff --git a/client/options_test.go b/client/options_test.go
index 53815f3..902b39b 100644
--- a/client/options_test.go
+++ b/client/options_test.go
@@ -3,7 +3,9 @@ package client
 import (
 	"crypto/tls"
 	"testing"
+	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -102,3 +104,14 @@ func TestWithEndpoint(t *testing.T) {
 		})
 	}
 }
+
+func TestWithReconnect(t *testing.T) {
+	timeout := 2 * time.Second
+	opts := &options{}
+	fn := WithReconnect(timeout, &backoff.ZeroBackOff{})
+	err := fn(opts)
+	require.NoError(t, err)
+	assert.Equal(t, timeout, opts.timeout)
+	assert.Equal(t, true, opts.reconnect)
+	assert.Equal(t, &backoff.ZeroBackOff{}, opts.backoff)
+}
diff --git a/server/server_integration_test.go b/server/server_integration_test.go
index 6f77e34..b67c8dc 100644
--- a/server/server_integration_test.go
+++ b/server/server_integration_test.go
@@ -226,7 +226,7 @@ func TestClientServerMonitor(t *testing.T) {
 	require.NotEmpty(t, reply[0].UUID.GoUUID)
 	ovsRow.UUID = reply[0].UUID.GoUUID
 
-	err = ovs.MonitorAll("test")
+	_, err = ovs.MonitorAll()
 	require.Nil(t, err)
 	require.Eventually(t, func() bool {
 		seenMutex.RLock()
diff --git a/test/ovs/ovs_integration_test.go b/test/ovs/ovs_integration_test.go
index 34066bb..f2583d5 100644
--- a/test/ovs/ovs_integration_test.go
+++ b/test/ovs/ovs_integration_test.go
@@ -4,11 +4,14 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"strings"
 	"testing"
 	"time"
 
+	"github.com/cenkalti/backoff/v4"
 	"github.com/ory/dockertest/v3"
 	"github.com/ory/dockertest/v3/docker"
+	"github.com/ovn-org/libovsdb/cache"
 	"github.com/ovn-org/libovsdb/client"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
@@ -63,7 +66,10 @@ func (suite *OVSIntegrationSuite) SetupSuite() {
 		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
 		defer cancel()
 		endpoint := fmt.Sprintf("tcp::%s", suite.resource.GetPort("6640/tcp"))
-		ovs, err := client.NewOVSDBClient(defDB, client.WithEndpoint(endpoint))
+		ovs, err := client.NewOVSDBClient(
+			defDB,
+			client.WithEndpoint(endpoint),
+		)
 		if err != nil {
 			return err
 		}
@@ -78,13 +84,13 @@ func (suite *OVSIntegrationSuite) SetupSuite() {
 
 	// give ovsdb-server some time to start up
 
-	err = suite.client.MonitorAll(nil)
+	_, err = suite.client.MonitorAll()
 	require.NoError(suite.T(), err)
 }
 
 func (suite *OVSIntegrationSuite) TearDownSuite() {
 	if suite.client != nil {
-		suite.client.Disconnect()
+		suite.client.Close()
 		suite.client = nil
 	}
 	err := suite.pool.Purge(suite.resource)
@@ -118,11 +124,29 @@ var defDB, _ = model.NewDBModel("Open_vSwitch", map[string]model.Model{
 	"Open_vSwitch": &ovsType{},
 	"Bridge":       &bridgeType{}})
 
-func (suite *OVSIntegrationSuite) TestConnectReconnectIntegration() {
+func (suite *OVSIntegrationSuite) TestConnectReconnect() {
 	assert.True(suite.T(), suite.client.Connected())
 	err := suite.client.Echo()
 	require.NoError(suite.T(), err)
 
+	bridgeName := "br-discoreco"
+	brChan := make(chan *bridgeType)
+	suite.client.Cache().AddEventHandler(&cache.EventHandlerFuncs{
+		AddFunc: func(table string, model model.Model) {
+			br, ok := model.(*bridgeType)
+			if !ok {
+				return
+			}
+			if br.Name == bridgeName {
+				brChan <- br
+			}
+		},
+	})
+
+	bridgeUUID, err := suite.createBridge(bridgeName)
+	require.NoError(suite.T(), err)
+	br := <-brChan
+
 	// 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()
@@ -160,93 +184,175 @@ func (suite *OVSIntegrationSuite) TestConnectReconnectIntegration() {
 	err = suite.client.Connect(ctx)
 	require.NoError(suite.T(), err)
 
+	br = &bridgeType{
+		UUID: bridgeUUID,
+	}
+
+	// assert cache has been purged
+	err = suite.client.Get(br)
+	require.Error(suite.T(), err, client.ErrNotFound)
+
 	err = suite.client.Echo()
 	assert.NoError(suite.T(), err)
 
-	err = suite.client.MonitorAll(nil)
+	_, err = suite.client.MonitorAll()
 	require.NoError(suite.T(), err)
+
+	// assert cache has been re-populated
+	require.Eventually(suite.T(), func() bool {
+		err := suite.client.Get(br)
+		return err == nil
+	}, 2*time.Second, 500*time.Millisecond)
+
 }
-func (suite *OVSIntegrationSuite) TestInsertTransactIntegration() {
-	bridgeName := "gopher-br7"
 
-	// NamedUUID is used to add multiple related Operations in a single Transact operation
-	namedUUID := "gopher"
-	br := bridgeType{
-		UUID: namedUUID,
-		Name: bridgeName,
-		ExternalIds: map[string]string{
-			"go":     "awesome",
-			"docker": "made-for-each-other",
+func (suite *OVSIntegrationSuite) TestWithReconnect() {
+	assert.Equal(suite.T(), true, suite.client.Connected())
+	err := suite.client.Echo()
+	require.NoError(suite.T(), err)
+
+	// Disconnect client
+	suite.client.Disconnect()
+
+	require.Eventually(suite.T(), func() bool {
+		return !suite.client.Connected()
+	}, 5*time.Second, 1*time.Second)
+
+	// Reconfigure
+	err = suite.client.SetOption(
+		client.WithReconnect(2*time.Second, backoff.NewExponentialBackOff()),
+	)
+	require.NoError(suite.T(), err)
+
+	// Connect (again)
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	err = suite.client.Connect(ctx)
+	require.NoError(suite.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 = suite.client.Connect(ctx)
+	require.NoError(suite.T(), err)
+
+	// check the connection is working
+	err = suite.client.Echo()
+	require.NoError(suite.T(), err)
+
+	// check the cache is purged
+	require.True(suite.T(), suite.client.Cache().Table("Bridge").Len() == 0)
+
+	// set up the monitor again
+	_, err = suite.client.MonitorAll()
+	require.NoError(suite.T(), err)
+
+	// add a bridge and verify our handler gets called
+	bridgeName := "recon-b4"
+	brChan := make(chan *bridgeType)
+	suite.client.Cache().AddEventHandler(&cache.EventHandlerFuncs{
+		AddFunc: func(table string, model model.Model) {
+			br, ok := model.(*bridgeType)
+			if !ok {
+				return
+			}
+			if strings.HasPrefix(br.Name, "recon-") {
+				brChan <- br
+			}
 		},
-	}
+	})
 
-	insertOp, err := suite.client.Create(&br)
+	_, err = suite.createBridge(bridgeName)
 	require.NoError(suite.T(), err)
+	br := <-brChan
+	require.Equal(suite.T(), bridgeName, br.Name)
 
-	// Inserting a Bridge row in Bridge table requires mutating the open_vswitch table.
-	ovsRow := ovsType{}
-	mutateOp, err := suite.client.WhereCache(func(*ovsType) bool { return true }).
-		Mutate(&ovsRow, model.Mutation{
-			Field:   &ovsRow.Bridges,
-			Mutator: ovsdb.MutateOperationInsert,
-			Value:   []string{namedUUID},
-		})
+	// trigger reconnect
+	suite.client.Disconnect()
+
+	// check that we are automatically reconnected
+	require.Eventually(suite.T(), func() bool {
+		return suite.client.Connected()
+	}, 2*time.Second, 500*time.Millisecond)
+
+	err = suite.client.Echo()
 	require.NoError(suite.T(), err)
 
-	operations := append(insertOp, mutateOp...)
-	reply, err := suite.client.Transact(operations...)
+	// check our original bridge is in the cache
+	err = suite.client.Get(br)
 	require.NoError(suite.T(), err)
 
-	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
-	if err != nil {
-		for _, oe := range operationErrs {
-			suite.T().Error(oe)
+	// create a new bridge to ensure the monitor and cache handler is still working
+	bridgeName = "recon-after"
+	_, err = suite.createBridge(bridgeName)
+	require.NoError(suite.T(), err)
+
+LOOP:
+	for {
+		select {
+		case <-time.After(2 * time.Second):
+			suite.T().Fatal("timed out waiting for bridge")
+		case b := <-brChan:
+			if b.Name == bridgeName {
+				break LOOP
+			}
 		}
-		suite.T().Fatal(err)
 	}
-}
 
-func (suite *OVSIntegrationSuite) TestInsertAndDeleteTransactIntegration() {
-	bridgeName := "gopher-br5"
-	namedUUID := "gopher"
-	br := bridgeType{
-		UUID: namedUUID,
-		Name: bridgeName,
-		ExternalIds: map[string]string{
-			"go":     "awesome",
-			"docker": "made-for-each-other",
-		},
+	// set up a disconnect notification
+	disconnectNotification := suite.client.DisconnectNotify()
+	notified := make(chan struct{})
+	ready := make(chan struct{})
+
+	go func() {
+		ready <- struct{}{}
+		<-disconnectNotification
+		notified <- struct{}{}
+	}()
+
+	<-ready
+	// close the connection
+	suite.client.Close()
+
+	select {
+	case <-notified:
+		// got notification
+	case <-time.After(5 * time.Second):
+		suite.T().Fatal("expected a disconnect notification but didn't receive one")
 	}
 
-	insertOp, err := suite.client.Create(&br)
+	assert.Equal(suite.T(), false, suite.client.Connected())
+
+	err = suite.client.Echo()
+	require.EqualError(suite.T(), err, client.ErrNotConnected.Error())
+
+	ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	err = suite.client.Connect(ctx)
 	require.NoError(suite.T(), err)
 
-	// Inserting a Bridge row in Bridge table requires mutating the open_vswitch table.
-	ovsRow := ovsType{}
-	mutateOp, err := suite.client.WhereCache(func(*ovsType) bool { return true }).
-		Mutate(&ovsRow, model.Mutation{
-			Field:   &ovsRow.Bridges,
-			Mutator: ovsdb.MutateOperationInsert,
-			Value:   []string{namedUUID},
-		})
+	err = suite.client.Echo()
+	assert.NoError(suite.T(), err)
+
+	_, err = suite.client.MonitorAll()
 	require.NoError(suite.T(), err)
+}
 
-	operations := append(insertOp, mutateOp...)
-	reply, err := suite.client.Transact(operations...)
+func (suite *OVSIntegrationSuite) TestInsertTransactIntegration() {
+	bridgeName := "gopher-br7"
+	_, err := suite.createBridge(bridgeName)
 	require.NoError(suite.T(), err)
+}
 
-	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
-	if err != nil {
-		for _, oe := range operationErrs {
-			suite.T().Error(oe)
-		}
-		suite.T().Fatal(err)
-	}
-	bridgeUUID := reply[0].UUID.GoUUID
+func (suite *OVSIntegrationSuite) TestInsertAndDeleteTransactIntegration() {
+	bridgeName := "gopher-br5"
+	bridgeUUID, err := suite.createBridge(bridgeName)
+	require.NoError(suite.T(), err)
 
 	deleteOp, err := suite.client.Where(&bridgeType{Name: bridgeName}).Delete()
 	require.NoError(suite.T(), err)
 
+	ovsRow := ovsType{}
 	delMutateOp, err := suite.client.WhereCache(func(*ovsType) bool { return true }).
 		Mutate(&ovsRow, model.Mutation{
 			Field:   &ovsRow.Bridges,
@@ -315,15 +421,13 @@ func (suite *OVSIntegrationSuite) TestColumnSchemaValidationIntegration() {
 }
 
 func (suite *OVSIntegrationSuite) TestMonitorCancelIntegration() {
-	monitorID := "f1b2ca48-aad7-11e7-abc4-cec278b6b50a"
-
 	requests := make(map[string]ovsdb.MonitorRequest)
 	requests["Bridge"] = ovsdb.MonitorRequest{
 		Columns: []string{"name"},
 		Select:  ovsdb.NewDefaultMonitorSelect(),
 	}
 
-	err := suite.client.Monitor(monitorID,
+	monitorID, err := suite.client.Monitor(
 		suite.client.NewTableMonitor(&ovsType{}),
 		suite.client.NewTableMonitor(&bridgeType{}),
 	)
@@ -334,11 +438,20 @@ func (suite *OVSIntegrationSuite) TestMonitorCancelIntegration() {
 }
 
 func (suite *OVSIntegrationSuite) TestInsertDuplicateTransactIntegration() {
+	_, err := suite.createBridge("br-dup")
+	require.NoError(suite.T(), err)
+
+	_, err = suite.createBridge("br-dup")
+	assert.Error(suite.T(), err)
+	assert.IsType(suite.T(), &ovsdb.ConstraintViolation{}, err)
+}
+
+func (suite *OVSIntegrationSuite) createBridge(bridgeName string) (string, error) {
 	// NamedUUID is used to add multiple related Operations in a single Transact operation
 	namedUUID := "gopher"
 	br := bridgeType{
 		UUID: namedUUID,
-		Name: "br-dup",
+		Name: bridgeName,
 		ExternalIds: map[string]string{
 			"go":     "awesome",
 			"docker": "made-for-each-other",
@@ -362,18 +475,6 @@ func (suite *OVSIntegrationSuite) TestInsertDuplicateTransactIntegration() {
 	reply, err := suite.client.Transact(operations...)
 	require.NoError(suite.T(), err)
 
-	operationErrs, err := ovsdb.CheckOperationResults(reply, operations)
-	if err != nil {
-		for _, oe := range operationErrs {
-			suite.T().Error(oe)
-		}
-		suite.T().Fatal(err)
-	}
-
-	reply2, err := suite.client.Transact(operations...)
-	require.NoError(suite.T(), err)
-
-	_, err = ovsdb.CheckOperationResults(reply2, operations)
-	assert.Error(suite.T(), err)
-	assert.IsType(suite.T(), &ovsdb.ConstraintViolation{}, err)
+	_, err = ovsdb.CheckOperationResults(reply, operations)
+	return reply[0].UUID.GoUUID, err
 }