Repository: ovn-org/libovsdb
Pull request #166 (2021-06-16 22:07:22)

Make Connect() Part of the Client. Make Client an interface.
There are a couple of big API changes here.

1. `client.Connect` is no longer the way to create a client. You now do this with `client.NewOVSDBClient`
1. `Register` has been removed, as we there isn't really anything useful for users to register a handler for. The cache takes care of update notificaitons and has it's own handlers for add/update/delete, lock/stolen aren't implemented, and we'd be better providing a channel for disconnects. As of now though, I've opted to just return errors when disconnected.
1. `Options` have been moved to `NewOVSDBClient` which will allows for the client to be customized, not just the parameters of `Connect()`
1. `NewOVSDBClient` returns an interface, this allows for adding a `FakeOVSDBClient` in future for testing purposes (i.e it would could use some of the `server` logic to handle transactions without having to worry about actually creating a server).

Minor changes:
1. There is a `connected bool` which tracks the state of the connection. This is used to allow methods to return `ErrNotConnected` if the server is unavailable
1. The `Disconnect` logic has been cleaned up to ensure the we don't have races when clearing the connection and then reconnecting
1. Adds `SetOption` to the `Client` to allow the user to change any options *after* the client is connected. This would only take effect if the client was disconnected and then reconnected again.

The motivation for this is to gracefully handle connection issues.

These can easily be detected if an RPC method returns `ErrNotConnected`, to which the user may wish to call `Connect()`, `Monitor()` etc... again to re-establish their connection.

`SetOption` allows for changing connection parameters when we're disconnected. E.g to change TLS certs.

coveralls: ## Pull Request Test Coverage Report for [Build 944460690](https://coveralls.io/builds/40650522)

* **110** of **141**   **(78.01%)**  changed or added relevant lines in **1** file are covered.
* **10** unchanged lines in **1** file lost coverage.
* Overall coverage increased (+**0.01%**) to **73.743%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [client/client.go](https://coveralls.io/builds/40650522/source?filename=client%2Fclient.go#L85) | 110 | 141 | 78.01%
<!-- | **Total:** | **110** | **141** | **78.01%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [client/client.go](https://coveralls.io/builds/40650522/source?filename=client%2Fclient.go#L147) | 10 | 68.73% |
<!-- | **Total:** | **10** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/40650522/badge)](https://coveralls.io/builds/40650522) |
| :-- | --: |
| Change from base [Build 944298636](https://coveralls.io/builds/40649004): |  0.01% |
| Covered Lines: | 2977 |
| Relevant Lines: | 4037 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/cache/cache.go b/cache/cache.go
index a457383..6bc800d 100644
--- a/cache/cache.go
+++ b/cache/cache.go
@@ -28,7 +28,7 @@ const (
 type IndexExistsError struct {
 	table    string
 	value    interface{}
-	index    index
+	column   string
 	new      string
 	existing string
 }
@@ -37,15 +37,15 @@ func (i *IndexExistsError) Error() string {
 	return fmt.Sprintf("operation would cause rows in the \"%s\" table to have identical values (%v) for index on column \"%s\". First row, with UUID %s, was inserted by this transaction. Second row, with UUID %s, existed in the database before this operation and was not modified",
 		i.table,
 		i.value,
-		i.index,
+		i.column,
 		i.new,
 		i.existing,
 	)
 }
 
-func NewIndexExistsError(table string, value interface{}, index index, new, existing string) *IndexExistsError {
+func NewIndexExistsError(table string, value interface{}, column, new, existing string) *IndexExistsError {
 	return &IndexExistsError{
-		table, value, index, new, existing,
+		table, value, column, new, existing,
 	}
 }
 
@@ -53,20 +53,7 @@ func NewIndexExistsError(table string, value interface{}, index index, new, exis
 type valueToUUID map[interface{}]string
 
 // map of column name(s) to a unique values, to UUIDs
-type columnToValue map[index]valueToUUID
-
-// index is the type used to implement multiple cache indexes
-type index string
-
-// columns returns the columns that conform the index
-func (i index) columns() []string {
-	return strings.Split(string(i), columnDelimiter)
-}
-
-// newIndex builds a index from a list of columns
-func newIndex(columns ...string) index {
-	return index(strings.Join(columns, columnDelimiter))
-}
+type columnToValue map[string]valueToUUID
 
 // RowCache is a collections of Models hashed by UUID
 type RowCache struct {
@@ -112,18 +99,27 @@ func (r *RowCache) create(uuid string, m model.Model) error {
 	}
 	newIndexes := newColumnToValue(r.schema.Indexes)
 	var errs []error
-	for index := range r.indexes {
-
-		val, err := valueFromIndex(info, index)
-
-		if err != nil {
-			return err
+	for columnStr := range r.indexes {
+		columns := strings.Split(columnStr, columnDelimiter)
+		var val interface{}
+		var err error
+		if len(columns) > 1 {
+			val, err = hashColumnValues(info, columns)
+			if err != nil {
+				return err
+			}
+		} else {
+			column := columns[0]
+			val, err = info.FieldByColumn(column)
+			if err != nil {
+				return err
+			}
 		}
-		if existing, ok := r.indexes[index][val]; ok {
+		if existing, ok := r.indexes[columnStr][val]; ok {
 			errs = append(errs,
-				NewIndexExistsError(r.name, val, index, uuid, existing))
+				NewIndexExistsError(r.name, val, columnStr, uuid, existing))
 		}
-		newIndexes[index][val] = uuid
+		newIndexes[columnStr][val] = uuid
 	}
 	if len(errs) != 0 {
 		return fmt.Errorf("%v", errs)
@@ -161,17 +157,31 @@ func (r *RowCache) update(uuid string, m model.Model) error {
 	newIndexes := newColumnToValue(r.schema.Indexes)
 	oldIndexes := newColumnToValue(r.schema.Indexes)
 	var errs []error
-	for index := range r.indexes {
+	for columnStr := range r.indexes {
+		columns := strings.Split(columnStr, columnDelimiter)
+		var oldVal interface{}
+		var newVal interface{}
 		var err error
-		oldVal, err := valueFromIndex(oldInfo, index)
-		if err != nil {
-			return err
-		}
-		newVal, err := valueFromIndex(newInfo, index)
-		if err != nil {
-			return err
+		if len(columns) > 1 {
+			oldVal, err = hashColumnValues(oldInfo, columns)
+			if err != nil {
+				return err
+			}
+			newVal, err = hashColumnValues(newInfo, columns)
+			if err != nil {
+				return err
+			}
+		} else {
+			column := columns[0]
+			oldVal, err = oldInfo.FieldByColumn(column)
+			if err != nil {
+				return err
+			}
+			newVal, err = newInfo.FieldByColumn(column)
+			if err != nil {
+				return err
+			}
 		}
-
 		// if old and new values are the same, don't worry
 		if oldVal == newVal {
 			continue
@@ -179,17 +189,17 @@ func (r *RowCache) update(uuid string, m model.Model) error {
 		// old and new values are NOT the same
 
 		// check that there are no conflicts
-		if conflict, ok := r.indexes[index][newVal]; ok && conflict != uuid {
+		if conflict, ok := r.indexes[columnStr][newVal]; ok && conflict != uuid {
 			errs = append(errs, NewIndexExistsError(
 				r.name,
 				newVal,
-				index,
+				columnStr,
 				uuid,
 				conflict,
 			))
 		}
-		newIndexes[index][newVal] = uuid
-		oldIndexes[index][oldVal] = ""
+		newIndexes[columnStr][newVal] = uuid
+		oldIndexes[columnStr][oldVal] = ""
 	}
 	if len(errs) > 0 {
 		return fmt.Errorf("%+v", errs)
@@ -226,12 +236,12 @@ func (r *RowCache) delete(uuid string) error {
 	if err != nil {
 		return err
 	}
-	for index := range r.indexes {
-		oldVal, err := valueFromIndex(oldInfo, index)
+	for column := range r.indexes {
+		oldVal, err := oldInfo.FieldByColumn(column)
 		if err != nil {
 			return err
 		}
-		delete(r.indexes[index], oldVal)
+		delete(r.indexes[column], oldVal)
 	}
 	delete(r.cache, uuid)
 	return nil
@@ -255,12 +265,12 @@ func (r *RowCache) Len() int {
 	return len(r.cache)
 }
 
-func (r *RowCache) Index(columns ...string) (map[interface{}]string, error) {
+func (r *RowCache) Index(column string) (map[interface{}]string, error) {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
-	index, ok := r.indexes[newIndex(columns...)]
+	index, ok := r.indexes[column]
 	if !ok {
-		return nil, fmt.Errorf("%s is not an index", index)
+		return nil, fmt.Errorf("%s is not an index", column)
 	}
 	return index, nil
 }
@@ -303,11 +313,14 @@ func (e *EventHandlerFuncs) OnDelete(table string, row model.Model) {
 
 // TableCache contains a collection of RowCaches, hashed by name,
 // and an array of EventHandlers that respond to cache updates
+// It implements the ovsdb.NotifcationHandler interface so it may
+// handle update notifications
 type TableCache struct {
 	cache          map[string]*RowCache
 	eventProcessor *eventProcessor
 	mapper         *mapper.Mapper
 	dbModel        *model.DBModel
+	ovsdb.NotificationHandler
 }
 
 // Data is the type for data that can be prepoulated in the cache
@@ -429,7 +442,11 @@ func (t *TableCache) Populate(tableUpdates ovsdb.TableUpdates) {
 						if err := tCache.update(uuid, newModel); err != nil {
 							panic(err)
 						}
-						t.eventProcessor.AddEvent(updateEvent, table, existing, newModel)
+						oldModel, err := t.CreateModel(table, row.Old, uuid)
+						if err != nil {
+							panic(err)
+						}
+						t.eventProcessor.AddEvent(updateEvent, table, oldModel, newModel)
 					}
 					// no diff
 					continue
@@ -483,9 +500,9 @@ func newColumnToValue(schemaIndexes [][]string) columnToValue {
 	// columns whose values, taken together within any given row, must be
 	// unique within the table". We'll store the column names, separated by comma
 	// as we'll assuume (RFC is not clear), that comma isn't valid in a <id>
-	var indexes []index
+	var indexes []string
 	for i := range schemaIndexes {
-		indexes = append(indexes, newIndex(schemaIndexes[i]...))
+		indexes = append(indexes, strings.Join(schemaIndexes[i], columnDelimiter))
 	}
 	c := make(columnToValue)
 	for _, index := range indexes {
@@ -603,28 +620,20 @@ func (t *TableCache) CreateModel(tableName string, row *ovsdb.Row, uuid string)
 	return model, nil
 }
 
-func valueFromIndex(info *mapper.Info, index index) (interface{}, error) {
-	columns := index.columns()
-	if len(columns) > 1 {
-		var buf bytes.Buffer
-		enc := gob.NewEncoder(&buf)
-		for _, column := range columns {
-			val, err := info.FieldByColumn(column)
-			if err != nil {
-				return "", err
-			}
-			err = enc.Encode(val)
-			if err != nil {
-				return "", err
-			}
+func hashColumnValues(info *mapper.Info, columns []string) (string, error) {
+	var buf bytes.Buffer
+	enc := gob.NewEncoder(&buf)
+	for _, column := range columns {
+		val, err := info.FieldByColumn(column)
+		if err != nil {
+			return "", err
+		}
+		err = enc.Encode(val)
+		if err != nil {
+			return "", err
 		}
-		h := sha256.New()
-		val := hex.EncodeToString(h.Sum(buf.Bytes()))
-		return val, nil
-	}
-	val, err := info.FieldByColumn(columns[0])
-	if err != nil {
-		return nil, err
 	}
-	return val, err
+	h := sha256.New()
+	val := hex.EncodeToString(h.Sum(buf.Bytes()))
+	return val, nil
 }
diff --git a/client/client.go b/client/client.go
index ad1d92d..d937c77 100644
--- a/client/client.go
+++ b/client/client.go
@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"net"
 	"net/url"
@@ -19,28 +20,6 @@ import (
 	"github.com/ovn-org/libovsdb/ovsdb"
 )
 
-// OvsdbClient is an OVSDB client
-type OvsdbClient struct {
-	rpcClient     *rpc2.Client
-	dbModel       *model.DBModel
-	Schema        ovsdb.DatabaseSchema
-	handlers      []ovsdb.NotificationHandler
-	handlersMutex *sync.Mutex
-	Cache         *cache.TableCache
-	stopCh        chan struct{}
-	api           API
-}
-
-func newOvsdbClient(dbModel *model.DBModel) *OvsdbClient {
-	// Cache initialization is delayed because we first need to obtain the schema
-	ovs := &OvsdbClient{
-		dbModel:       dbModel,
-		handlersMutex: &sync.Mutex{},
-		stopCh:        make(chan struct{}),
-	}
-	return ovs
-}
-
 // Constants defined for libovsdb
 const (
 	SSL  = "ssl"
@@ -48,24 +27,81 @@ const (
 	UNIX = "unix"
 )
 
-// Connect to an OVSDB Server using the provided endpoint in OVSDB Connection Format
-// For more details, see the ovsdb(7) man page
+// ErrNotConnected is an error returned when the client is not connected
+var ErrNotConnected = errors.New("not connected")
+
+// Client represents an OVSDB Client Connection
+// It provides all the necessary functionality to Connect to a server,
+// perform transactions, and build your own replica of the database with
+// Monitor or MonitorAll. It also provides a Cache that is populated from OVSDB
+// update notifications.
+type Client interface {
+	Connect(context.Context) error
+	Disconnect()
+	Schema() *ovsdb.DatabaseSchema
+	Cache() *cache.TableCache
+	SetOption(Option) error
+	Echo() error
+	Transact(...ovsdb.Operation) ([]ovsdb.OperationResult, error)
+	Monitor(jsonContext interface{}, t ...TableMonitor) error
+	MonitorAll(jsonContext interface{}) error
+	MonitorCancel(jsonContext interface{}) 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{}
+	connected bool
+	api       API
+	mutex     sync.Mutex
+}
+
+// NewOVSDBClient creates a new OVSDB Client with the provided
+// database model. The client 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 NewOVSDBClient(databaseModel *model.DBModel, opts ...Option) (Client, error) {
+	return newOVSDBClient(databaseModel, opts...)
+}
+
+// newOVSDBClient creates a new ovsdbClient
+func newOVSDBClient(databaseModel *model.DBModel, opts ...Option) (*ovsdbClient, error) {
+	ovs := &ovsdbClient{
+		dbModel: databaseModel,
+	}
+	var err error
+	ovs.options, err = newOptions(opts...)
+	if err != nil {
+		return nil, err
+	}
+	return ovs, nil
+}
+
+// Connect opens a connection to an OVSDB Server using the
+// endpoint provided when the Client was created.
 // 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 Connect(ctx context.Context, database *model.DBModel, opts ...Option) (*OvsdbClient, error) {
+func (o *ovsdbClient) Connect(ctx context.Context) error {
+	o.mutex.Lock()
+	defer o.mutex.Unlock()
+	if o.connected {
+		return nil
+	}
 	var c net.Conn
 	var dialer net.Dialer
 	var err error
 	var u *url.URL
-
-	options, err := newOptions(opts...)
-	if err != nil {
-		return nil, err
-	}
-
-	for _, endpoint := range options.endpoints {
+	connected := false
+	for _, endpoint := range o.options.endpoints {
 		if u, err = url.Parse(endpoint); err != nil {
-			return nil, err
+			return err
 		}
 		switch u.Scheme {
 		case UNIX:
@@ -74,123 +110,122 @@ func Connect(ctx context.Context, database *model.DBModel, opts ...Option) (*Ovs
 			c, err = dialer.DialContext(ctx, u.Scheme, u.Opaque)
 		case SSL:
 			dialer := tls.Dialer{
-				Config: options.tlsConfig,
+				Config: o.options.tlsConfig,
 			}
 			c, err = dialer.DialContext(ctx, "tcp", u.Opaque)
 		default:
 			err = fmt.Errorf("unknown network protocol %s", u.Scheme)
 		}
 		if err == nil {
-			return newRPC2Client(c, database)
+			connected = true
+			break
 		}
 	}
-	return nil, fmt.Errorf("failed to connect to endpoints %q: %v", options.endpoints, err)
+	if !connected {
+		// FIXME: This only emits the error from the last attempted connection
+		return fmt.Errorf("failed to connect to endpoints %q: %v", o.options.endpoints, err)
+	}
+	if err := o.createRPC2Client(c); err != nil {
+		return err
+	}
+	o.connected = true
+	return nil
 }
 
-func newRPC2Client(conn net.Conn, database *model.DBModel) (*OvsdbClient, error) {
-	ovs := newOvsdbClient(database)
-	ovs.rpcClient = rpc2.NewClientWithCodec(jsonrpc.NewJSONCodec(conn))
-	ovs.rpcClient.SetBlocking(true)
-	ovs.rpcClient.Handle("echo", func(_ *rpc2.Client, args []interface{}, reply *[]interface{}) error {
-		return ovs.echo(args, reply)
+// createRPC2Client creates an rpcClient using the provided connection
+// It is also responsible for setting up go routines for handling disconnect notification
+// and cache population
+func (o *ovsdbClient) createRPC2Client(conn net.Conn) error {
+	o.stopCh = make(chan struct{})
+	o.rpcClient = rpc2.NewClientWithCodec(jsonrpc.NewJSONCodec(conn))
+	o.rpcClient.SetBlocking(true)
+	o.rpcClient.Handle("echo", func(_ *rpc2.Client, args []interface{}, reply *[]interface{}) error {
+		return o.echo(args, reply)
 	})
-	ovs.rpcClient.Handle("update", func(_ *rpc2.Client, args []json.RawMessage, reply *[]interface{}) error {
-		return ovs.update(args, reply)
+	o.rpcClient.Handle("update", func(_ *rpc2.Client, args []json.RawMessage, reply *[]interface{}) error {
+		return o.update(args, reply)
 	})
-	go ovs.rpcClient.Run()
-	go ovs.handleDisconnectNotification()
+	go o.rpcClient.Run()
 
-	dbs, err := ovs.ListDbs()
+	dbs, err := o.listDbs()
 	if err != nil {
-		ovs.rpcClient.Close()
-		return nil, err
+		o.rpcClient.Close()
+		return err
 	}
 
 	found := false
 	for _, db := range dbs {
-		if db == database.Name() {
+		if db == o.dbModel.Name() {
 			found = true
 			break
 		}
 	}
 	if !found {
-		ovs.rpcClient.Close()
-		return nil, fmt.Errorf("target database not found")
+		o.rpcClient.Close()
+		return fmt.Errorf("target database not found")
 	}
 
-	schema, err := ovs.GetSchema(database.Name())
-	errors := database.Validate(schema)
+	schema, err := o.getSchema(o.dbModel.Name())
+	errors := o.dbModel.Validate(schema)
 	if len(errors) > 0 {
 		var combined []string
 		for _, err := range errors {
 			combined = append(combined, err.Error())
 		}
-		return nil, fmt.Errorf("database validation error (%d): %s", len(errors),
+		return fmt.Errorf("database validation error (%d): %s", len(errors),
 			strings.Join(combined, ". "))
 	}
 
 	if err == nil {
-		ovs.Schema = *schema
-		if cache, err := cache.NewTableCache(schema, database, nil); err == nil {
-			ovs.Cache = cache
-			ovs.Register(ovs.Cache)
-			ovs.api = newAPI(ovs.Cache)
+		o.schema = schema
+		if cache, err := cache.NewTableCache(schema, o.dbModel, nil); err == nil {
+			o.cache = cache
+			o.api = newAPI(o.cache)
 		} else {
-			ovs.rpcClient.Close()
-			return nil, err
+			o.rpcClient.Close()
+			return err
 		}
 	} else {
-		ovs.rpcClient.Close()
-		return nil, err
+		o.rpcClient.Close()
+		return err
 	}
 
-	go ovs.Cache.Run(ovs.stopCh)
+	go o.cache.Run(o.stopCh)
+	go o.handleDisconnectNotification()
 
-	return ovs, nil
+	return nil
 }
 
-// Register registers the supplied NotificationHandler to receive OVSDB Notifications
-func (ovs *OvsdbClient) Register(handler ovsdb.NotificationHandler) {
-	ovs.handlersMutex.Lock()
-	defer ovs.handlersMutex.Unlock()
-	ovs.handlers = append(ovs.handlers, handler)
+// 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 {
+	return o.schema
 }
 
-//Get Handler by index
-func getHandlerIndex(handler ovsdb.NotificationHandler, handlers []ovsdb.NotificationHandler) (int, error) {
-	for i, h := range handlers {
-		if reflect.DeepEqual(h, handler) {
-			return i, nil
-		}
-	}
-	return -1, fmt.Errorf("handler not found")
+// Cache returns the TableCache that is populated from
+// 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 {
+	return o.cache
 }
 
-// Unregister the supplied NotificationHandler to not receive OVSDB Notifications anymore
-func (ovs *OvsdbClient) Unregister(handler ovsdb.NotificationHandler) error {
-	ovs.handlersMutex.Lock()
-	defer ovs.handlersMutex.Unlock()
-	i, err := getHandlerIndex(handler, ovs.handlers)
-	if err != nil {
-		return err
+// 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 {
+	if o.connected {
+		return fmt.Errorf("cannot set option when client is connected")
 	}
-	ovs.handlers = append(ovs.handlers[:i], ovs.handlers[i+1:]...)
-	return nil
+	return opt(o.options)
 }
 
 // RFC 7047 : Section 4.1.6 : Echo
-func (ovs *OvsdbClient) echo(args []interface{}, reply *[]interface{}) error {
+func (o *ovsdbClient) echo(args []interface{}, reply *[]interface{}) error {
 	*reply = args
-	ovs.handlersMutex.Lock()
-	defer ovs.handlersMutex.Unlock()
-	for _, handler := range ovs.handlers {
-		handler.Echo(nil)
-	}
 	return nil
 }
 
 // RFC 7047 : Update Notification Section 4.1.6
-func (ovs *OvsdbClient) update(args []json.RawMessage, reply *[]interface{}) error {
+func (o *ovsdbClient) update(args []json.RawMessage, reply *[]interface{}) error {
 	var value string
 	if len(args) > 2 {
 		return fmt.Errorf("update requires exactly 2 args")
@@ -205,50 +240,48 @@ func (ovs *OvsdbClient) update(args []json.RawMessage, reply *[]interface{}) err
 		return err
 	}
 	// Update the local DB cache with the tableUpdates
-	ovs.handlersMutex.Lock()
-	defer ovs.handlersMutex.Unlock()
-	for _, handler := range ovs.handlers {
-		handler.Update(value, updates)
-	}
+	o.cache.Update(value, updates)
 	*reply = []interface{}{}
 	return nil
 }
 
-// GetSchema returns the schema in use for the provided database name
+// getSchema returns the schema in use for the provided database name
 // RFC 7047 : get_schema
-func (ovs OvsdbClient) GetSchema(dbName string) (*ovsdb.DatabaseSchema, error) {
+func (o *ovsdbClient) getSchema(dbName string) (*ovsdb.DatabaseSchema, error) {
 	args := ovsdb.NewGetSchemaArgs(dbName)
 	var reply ovsdb.DatabaseSchema
-	err := ovs.rpcClient.Call("get_schema", args, &reply)
+	err := o.rpcClient.Call("get_schema", args, &reply)
 	if err != nil {
 		return nil, err
 	}
-	ovs.Schema = reply
 	return &reply, err
 }
 
-// ListDbs returns the list of databases on the server
+// listDbs returns the list of databases on the server
 // RFC 7047 : list_dbs
-func (ovs OvsdbClient) ListDbs() ([]string, error) {
+func (o *ovsdbClient) listDbs() ([]string, error) {
 	var dbs []string
-	err := ovs.rpcClient.Call("list_dbs", nil, &dbs)
+	err := o.rpcClient.Call("list_dbs", nil, &dbs)
 	if err != nil {
 		return nil, fmt.Errorf("listdbs failure - %v", err)
 	}
 	return dbs, err
 }
 
-// Transact performs the provided Operation's on the database
+// Transact performs the provided Operations on the database
 // RFC 7047 : transact
-func (ovs OvsdbClient) Transact(operation ...ovsdb.Operation) ([]ovsdb.OperationResult, error) {
+func (o *ovsdbClient) Transact(operation ...ovsdb.Operation) ([]ovsdb.OperationResult, error) {
+	if !o.connected {
+		return nil, ErrNotConnected
+	}
 	var reply []ovsdb.OperationResult
 
-	if ok := ovs.Schema.ValidateOperations(operation...); !ok {
+	if ok := o.schema.ValidateOperations(operation...); !ok {
 		return nil, fmt.Errorf("validation failed for the operation")
 	}
 
-	args := ovsdb.NewTransactArgs(ovs.Schema.Name, operation...)
-	err := ovs.rpcClient.Call("transact", args, &reply)
+	args := ovsdb.NewTransactArgs(o.schema.Name, operation...)
+	err := o.rpcClient.Call("transact", args, &reply)
 	if err != nil {
 		return nil, err
 	}
@@ -256,22 +289,28 @@ func (ovs OvsdbClient) Transact(operation ...ovsdb.Operation) ([]ovsdb.Operation
 }
 
 // MonitorAll is a convenience method to monitor every table/column
-func (ovs OvsdbClient) MonitorAll(jsonContext interface{}) error {
+func (o *ovsdbClient) MonitorAll(jsonContext interface{}) error {
+	if !o.connected {
+		return ErrNotConnected
+	}
 	var options []TableMonitor
-	for name := range ovs.dbModel.Types() {
+	for name := range o.dbModel.Types() {
 		options = append(options, TableMonitor{Table: name})
 	}
-	return ovs.Monitor(jsonContext, options...)
+	return o.Monitor(jsonContext, options...)
 }
 
 // MonitorCancel will request cancel a previously issued monitor request
 // RFC 7047 : monitor_cancel
-func (ovs OvsdbClient) MonitorCancel(jsonContext interface{}) error {
+func (o *ovsdbClient) MonitorCancel(jsonContext interface{}) error {
+	if !o.connected {
+		return ErrNotConnected
+	}
 	var reply ovsdb.OperationResult
 
 	args := ovsdb.NewMonitorCancelArgs(jsonContext)
 
-	err := ovs.rpcClient.Call("monitor_cancel", args, &reply)
+	err := o.rpcClient.Call("monitor_cancel", args, &reply)
 	if err != nil {
 		return err
 	}
@@ -292,8 +331,8 @@ type TableMonitor struct {
 	Error error
 }
 
-func (ovs *OvsdbClient) NewTableMonitor(m model.Model, fields ...interface{}) TableMonitor {
-	tableName := ovs.dbModel.FindTable(reflect.TypeOf(m))
+func (o *ovsdbClient) NewTableMonitor(m model.Model, fields ...interface{}) TableMonitor {
+	tableName := o.dbModel.FindTable(reflect.TypeOf(m))
 	if tableName == "" {
 		return TableMonitor{
 			Error: fmt.Errorf("object of type %s is not part of the DBModel", reflect.TypeOf(m)),
@@ -309,13 +348,16 @@ func (ovs *OvsdbClient) NewTableMonitor(m model.Model, fields ...interface{}) Ta
 // and populate the cache with them. Subsequent updates will be processed
 // by the Update Notifications
 // RFC 7047 : monitor
-func (ovs OvsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor) error {
+func (o *ovsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor) error {
+	if !o.connected {
+		return ErrNotConnected
+	}
 	if len(options) == 0 {
 		return fmt.Errorf("no monitor options provided")
 	}
 	var reply ovsdb.TableUpdates
-	mapper := mapper.NewMapper(&ovs.Schema)
-	typeMap := ovs.dbModel.Types()
+	mapper := mapper.NewMapper(o.schema)
+	typeMap := o.dbModel.Types()
 	requests := make(map[string]ovsdb.MonitorRequest)
 	for _, o := range options {
 		if o.Error != nil {
@@ -331,20 +373,23 @@ func (ovs OvsdbClient) Monitor(jsonContext interface{}, options ...TableMonitor)
 		}
 		requests[o.Table] = *request
 	}
-	args := ovsdb.NewMonitorArgs(ovs.Schema.Name, jsonContext, requests)
-	err := ovs.rpcClient.Call("monitor", args, &reply)
+	args := ovsdb.NewMonitorArgs(o.schema.Name, jsonContext, requests)
+	err := o.rpcClient.Call("monitor", args, &reply)
 	if err != nil {
 		return err
 	}
-	ovs.Cache.Populate(reply)
+	o.cache.Populate(reply)
 	return nil
 }
 
 // Echo tests the liveness of the OVSDB connetion
-func (ovs *OvsdbClient) Echo() error {
+func (o *ovsdbClient) Echo() error {
+	if !o.connected {
+		return ErrNotConnected
+	}
 	args := ovsdb.NewEchoArgs()
 	var reply []interface{}
-	err := ovs.rpcClient.Call("echo", args, &reply)
+	err := o.rpcClient.Call("echo", args, &reply)
 	if err != nil {
 		return err
 	}
@@ -354,59 +399,58 @@ func (ovs *OvsdbClient) Echo() error {
 	return nil
 }
 
-func (ovs *OvsdbClient) clearConnection() {
-	for _, handler := range ovs.handlers {
-		if handler != nil {
-			handler.Disconnected()
-		}
-	}
+func (o *ovsdbClient) handleDisconnectNotification() {
+	// this will block until Connect() has released the lock via defer
+	o.mutex.Lock()
+	// we continue to hold the lock until the client has disconnected
+	// this prevents another call to Connect() changing the rpcClient
+	// while we're still listening for disconnects
+	defer o.mutex.Unlock()
+	<-o.rpcClient.DisconnectNotify()
+	close(o.stopCh)
+	o.rpcClient = nil
+	o.cache = nil
 }
 
-func (ovs *OvsdbClient) handleDisconnectNotification() {
-	disconnected := ovs.rpcClient.DisconnectNotify()
-	<-disconnected
-	ovs.clearConnection()
-}
-
-// Disconnect will close the OVSDB connection
-func (ovs OvsdbClient) Disconnect() {
-	close(ovs.stopCh)
-	ovs.rpcClient.Close()
+// Disconnect will close the connection to the OVSDB server
+func (o *ovsdbClient) Disconnect() {
+	if !o.connected {
+		return
+	}
+	o.connected = false
+	o.rpcClient.Close()
 }
 
 // Client API interface wrapper functions
 // We add this wrapper to allow users to access the API directly on the
 // client object
 
-// Ensure client implements API
-var _ API = OvsdbClient{}
-
 //Get implements the API interface's Get function
-func (ovs OvsdbClient) Get(model model.Model) error {
-	return ovs.api.Get(model)
+func (o *ovsdbClient) Get(model model.Model) error {
+	return o.api.Get(model)
 }
 
 //Create implements the API interface's Create function
-func (ovs OvsdbClient) Create(models ...model.Model) ([]ovsdb.Operation, error) {
-	return ovs.api.Create(models...)
+func (o *ovsdbClient) Create(models ...model.Model) ([]ovsdb.Operation, error) {
+	return o.api.Create(models...)
 }
 
 //List implements the API interface's List function
-func (ovs OvsdbClient) List(result interface{}) error {
-	return ovs.api.List(result)
+func (o *ovsdbClient) List(result interface{}) error {
+	return o.api.List(result)
 }
 
 //Where implements the API interface's Where function
-func (ovs OvsdbClient) Where(m model.Model, conditions ...model.Condition) ConditionalAPI {
-	return ovs.api.Where(m, conditions...)
+func (o *ovsdbClient) Where(m model.Model, conditions ...model.Condition) ConditionalAPI {
+	return o.api.Where(m, conditions...)
 }
 
 //WhereAll implements the API interface's WhereAll function
-func (ovs OvsdbClient) WhereAll(m model.Model, conditions ...model.Condition) ConditionalAPI {
-	return ovs.api.WhereAll(m, conditions...)
+func (o *ovsdbClient) WhereAll(m model.Model, conditions ...model.Condition) ConditionalAPI {
+	return o.api.WhereAll(m, conditions...)
 }
 
 //WhereCache implements the API interface's WhereCache function
-func (ovs OvsdbClient) WhereCache(predicate interface{}) ConditionalAPI {
-	return ovs.api.WhereCache(predicate)
+func (o *ovsdbClient) WhereCache(predicate interface{}) ConditionalAPI {
+	return o.api.WhereCache(predicate)
 }
diff --git a/client/options.go b/client/options.go
index 27cc81c..d73f95d 100644
--- a/client/options.go
+++ b/client/options.go
@@ -42,7 +42,9 @@ func WithTLSConfig(cfg *tls.Config) Option {
 
 // WithEndpoint sets the endpoint to be used by the client
 // It can be used multiple times, and the first endpoint that
-// successfully connects will be used:
+// successfully connects will be used.
+// Endpoints are specified in OVSDB Connection Format
+// For more details, see the ovsdb(7) man page
 func WithEndpoint(endpoint string) Option {
 	return func(o *options) error {
 		ep, err := url.Parse(endpoint)
diff --git a/cmd/stress/stress.go b/cmd/stress/stress.go
index 88459c6..50b89c0 100644
--- a/cmd/stress/stress.go
+++ b/cmd/stress/stress.go
@@ -54,7 +54,11 @@ type result struct {
 }
 
 func cleanup(ctx context.Context) {
-	ovs, err := client.Connect(context.Background(), dbModel, client.WithEndpoint(*connection))
+	ovs, err := client.NewOVSDBClient(dbModel, client.WithEndpoint(*connection))
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = ovs.Connect(context.Background())
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -66,7 +70,7 @@ func cleanup(ctx context.Context) {
 
 	var rootUUID string
 	// Get root UUID
-	for _, uuid := range ovs.Cache.Table("Open_vSwitch").Rows() {
+	for _, uuid := range ovs.Cache().Table("Open_vSwitch").Rows() {
 		rootUUID = uuid
 		log.Printf("rootUUID is %v", rootUUID)
 	}
@@ -92,7 +96,11 @@ func run(ctx context.Context, resultsChan chan result, wg *sync.WaitGroup) {
 	ready := false
 	var rootUUID string
 
-	ovs, err := client.Connect(context.Background(), dbModel, client.WithEndpoint(*connection))
+	ovs, err := client.NewOVSDBClient(dbModel, client.WithEndpoint(*connection))
+	if err != nil {
+		log.Fatal(err)
+	}
+	err = ovs.Connect(context.Background())
 	if err != nil {
 		log.Fatal(err)
 	}
@@ -106,7 +114,7 @@ func run(ctx context.Context, resultsChan chan result, wg *sync.WaitGroup) {
 		bridgeCh[br.Name] = make(chan bool)
 	}
 
-	ovs.Cache.AddEventHandler(
+	ovs.Cache().AddEventHandler(
 		&cache.EventHandlerFuncs{
 			AddFunc: func(table string, model model.Model) {
 				if ready && table == "Bridge" {
@@ -133,7 +141,7 @@ func run(ctx context.Context, resultsChan chan result, wg *sync.WaitGroup) {
 	}
 
 	// Get root UUID
-	for _, uuid := range ovs.Cache.Table("Open_vSwitch").Rows() {
+	for _, uuid := range ovs.Cache().Table("Open_vSwitch").Rows() {
 		rootUUID = uuid
 		if *verbose {
 			fmt.Printf("rootUUID is %v\n", rootUUID)
@@ -157,7 +165,7 @@ func run(ctx context.Context, resultsChan chan result, wg *sync.WaitGroup) {
 	resultsChan <- result
 }
 
-func transact(ctx context.Context, ovs *client.OvsdbClient, operations []ovsdb.Operation) (bool, string) {
+func transact(ctx context.Context, ovs client.Client, operations []ovsdb.Operation) (bool, string) {
 	reply, err := ovs.Transact(operations...)
 	if err != nil {
 		return false, ""
@@ -168,7 +176,7 @@ func transact(ctx context.Context, ovs *client.OvsdbClient, operations []ovsdb.O
 	return true, reply[0].UUID.GoUUID
 }
 
-func deleteBridge(ctx context.Context, ovs *client.OvsdbClient, rootUUID string, bridge *bridgeType) {
+func deleteBridge(ctx context.Context, ovs client.Client, rootUUID string, bridge *bridgeType) {
 	log.Printf("deleting bridge %s", bridge.Name)
 	deleteOp, err := ovs.Where(bridge).Delete()
 	if err != nil {
@@ -204,7 +212,7 @@ func newBridge() bridgeType {
 	}
 }
 
-func createBridge(ctx context.Context, ovs *client.OvsdbClient, rootUUID string, bridge bridgeType) {
+func createBridge(ctx context.Context, ovs client.Client, rootUUID string, bridge bridgeType) {
 	insertOp, err := ovs.Create(&bridge)
 	if 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 a1e9868..7861394 100644
--- a/example/play_with_ovs/play_with_ovs.go
+++ b/example/play_with_ovs/play_with_ovs.go
@@ -25,7 +25,7 @@ var update chan model.Model
 var rootUUID string
 var connection = flag.String("ovsdb", "unix:/var/run/openvswitch/db.sock", "OVSDB connection string")
 
-func play(ovs *client.OvsdbClient) {
+func play(ovs client.Client) {
 	go processInput(ovs)
 	for model := range update {
 		bridge := model.(*Bridge)
@@ -46,7 +46,7 @@ func play(ovs *client.OvsdbClient) {
 	}
 }
 
-func createBridge(ovs *client.OvsdbClient, bridgeName string) {
+func createBridge(ovs client.Client, bridgeName string) {
 	bridge := Bridge{
 		UUID: "gopher",
 		Name: bridgeName,
@@ -79,7 +79,7 @@ func createBridge(ovs *client.OvsdbClient, bridgeName string) {
 	fmt.Println("Bridge Addition Successful : ", reply[0].UUID.GoUUID)
 }
 
-func processInput(ovs *client.OvsdbClient) {
+func processInput(ovs client.Client) {
 	for {
 		fmt.Printf("\n Enter a Bridge Name : ")
 		var bridgeName string
@@ -96,18 +96,23 @@ func main() {
 	quit = make(chan bool)
 	update = make(chan model.Model)
 
-	dbmodel, err := model.NewDBModel("Open_vSwitch",
+	dbModel, err := model.NewDBModel("Open_vSwitch",
 		map[string]model.Model{bridgeTable: &Bridge{}, ovsTable: &OpenvSwitch{}})
 	if err != nil {
 		log.Fatal("Unable to create DB model ", err)
 	}
 
-	ovs, err := client.Connect(context.Background(), dbmodel, client.WithEndpoint(*connection))
+	ovs, err := client.NewOVSDBClient(dbModel, client.WithEndpoint(*connection))
 	if err != nil {
-		log.Fatal("Unable to Connect ", err)
+		log.Fatal(err)
+	}
+	err = ovs.Connect(context.Background())
+	if err != nil {
+		log.Fatal(err)
 	}
+	defer ovs.Disconnect()
 
-	ovs.Cache.AddEventHandler(&cache.EventHandlerFuncs{
+	ovs.Cache().AddEventHandler(&cache.EventHandlerFuncs{
 		AddFunc: func(table string, model model.Model) {
 			if table == bridgeTable {
 				update <- model
@@ -122,7 +127,7 @@ func main() {
 		log.Fatal(err)
 	}
 
-	rootUUID = ovs.Cache.Table(ovsTable).Rows()[0]
+	rootUUID = ovs.Cache().Table(ovsTable).Rows()[0]
 
 	fmt.Println(`Silly game of stopping this app when a Bridge with name "stop" is monitored !`)
 	go play(ovs)
diff --git a/ovsdb/bindings.go b/ovsdb/bindings.go
index e0a714c..6e844ab 100644
--- a/ovsdb/bindings.go
+++ b/ovsdb/bindings.go
@@ -165,20 +165,6 @@ func OvsToNative(column *ColumnSchema, ovsElem interface{}) (interface{}, error)
 	}
 }
 
-// NativeToOvsAtomic returns the OVS type of the atomic native value
-func NativeToOvsAtomic(basicType string, nativeElem interface{}) (interface{}, error) {
-	naType := NativeTypeFromAtomic(basicType)
-	if reflect.TypeOf(nativeElem) != naType {
-		return nil, NewErrWrongType("NativeToOvsAtomic", naType.String(), nativeElem)
-	}
-	switch basicType {
-	case TypeUUID:
-		return UUID{GoUUID: nativeElem.(string)}, nil
-	default:
-		return nativeElem, nil
-	}
-}
-
 // NativeToOvs transforms an native type to a ovs type based on the column type information
 func NativeToOvs(column *ColumnSchema, rawElem interface{}) (interface{}, error) {
 	naType := NativeType(column)
@@ -211,21 +197,11 @@ func NativeToOvs(column *ColumnSchema, rawElem interface{}) (interface{}, error)
 		}
 		return ovsSet, nil
 	case TypeMap:
-		nativeMapVal := reflect.ValueOf(rawElem)
-		ovsMap := make(map[interface{}]interface{}, nativeMapVal.Len())
-		for _, key := range nativeMapVal.MapKeys() {
-			ovsKey, err := NativeToOvsAtomic(column.TypeObj.Key.Type, key.Interface())
-			if err != nil {
-				return nil, err
-			}
-			ovsVal, err := NativeToOvsAtomic(column.TypeObj.Value.Type, nativeMapVal.MapIndex(key).Interface())
-			if err != nil {
-				return nil, err
-			}
-			ovsMap[ovsKey] = ovsVal
+		ovsMap, err := NewOvsMap(rawElem)
+		if err != nil {
+			return nil, err
 		}
-		return OvsMap{GoMap: ovsMap}, nil
-
+		return ovsMap, nil
 	default:
 		panic(fmt.Sprintf("Unknown Type: %v", column.Type))
 	}
diff --git a/ovsdb/map.go b/ovsdb/map.go
index 78a309c..c571ff6 100644
--- a/ovsdb/map.go
+++ b/ovsdb/map.go
@@ -42,20 +42,7 @@ func (o *OvsMap) UnmarshalJSON(b []byte) (err error) {
 		innerSlice := oMap[1].([]interface{})
 		for _, val := range innerSlice {
 			f := val.([]interface{})
-			switch f[1].(type) {
-			case []interface{}:
-				vSet := f[1].([]interface{})
-				if len(vSet) != 2 || vSet[0] == "map" {
-					return &json.UnmarshalTypeError{Value: reflect.ValueOf(oMap).String(), Type: reflect.TypeOf(*o)}
-				}
-				goSlice, err := ovsSliceToGoNotation(vSet)
-				if err != nil {
-					return err
-				}
-				o.GoMap[f[0]] = goSlice
-			default:
-				o.GoMap[f[0]] = f[1]
-			}
+			o.GoMap[f[0]] = f[1]
 		}
 	}
 	return err


Test patch:
diff --git a/cache/cache_test.go b/cache/cache_test.go
index d2055ca..38088ea 100644
--- a/cache/cache_test.go
+++ b/cache/cache_test.go
@@ -226,7 +226,7 @@ func TestRowCacheCreateMultiIndex(t *testing.T) {
 				assert.Nil(t, err)
 				mapperInfo, err := mapper.NewInfo(tSchema, tt.model)
 				require.Nil(t, err)
-				h, err := valueFromIndex(mapperInfo, newIndex("foo", "bar"))
+				h, err := hashColumnValues(mapperInfo, []string{"foo", "bar"})
 				require.Nil(t, err)
 				assert.Equal(t, tt.uuid, rc.indexes["foo,bar"][h])
 			}
@@ -379,7 +379,7 @@ func TestRowCacheUpdateMultiIndex(t *testing.T) {
 				assert.Nil(t, err)
 				mapperInfo, err := mapper.NewInfo(tSchema, tt.model)
 				require.Nil(t, err)
-				h, err := valueFromIndex(mapperInfo, newIndex("foo", "bar"))
+				h, err := hashColumnValues(mapperInfo, []string{"foo", "bar"})
 				require.Nil(t, err)
 				assert.Equal(t, tt.uuid, rc.indexes["foo,bar"][h])
 			}
@@ -741,103 +741,3 @@ func TestEventProcessor_AddEvent(t *testing.T) {
 	// assert channel is empty
 	assert.Equal(t, 0, len(ep.events))
 }
-
-func TestIndex(t *testing.T) {
-	type indexTestModel struct {
-		UUID string `ovsdb:"_uuid"`
-		Foo  string `ovsdb:"foo"`
-		Bar  string `ovsdb:"bar"`
-		Baz  int    `ovsdb:"baz"`
-	}
-	db, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &indexTestModel{}})
-	assert.Nil(t, err)
-	var schema ovsdb.DatabaseSchema
-	err = json.Unmarshal([]byte(`
-		 {"name": "TestDB",
-		  "tables": {
-		    "Open_vSwitch": {
-			  "indexes": [["foo"], ["bar","baz"]],
-		      "columns": {
-		        "foo": {
-			  "type": "string"
-			},
-			"bar": {
-				"type": "string"
-			},
-			"baz": {
-				"type": "integer"
-			}
-		      }
-		    }
-		 }
-	     }
-	`), &schema)
-	assert.Nil(t, err)
-	tc, err := NewTableCache(&schema, db, nil)
-	assert.Nil(t, err)
-	obj := &indexTestModel{
-		UUID: "test1",
-		Foo:  "foo",
-		Bar:  "bar",
-		Baz:  42,
-	}
-	table := tc.Table("Open_vSwitch")
-
-	err = table.Create(obj.UUID, obj)
-	assert.Nil(t, err)
-	t.Run("Index by single column", func(t *testing.T) {
-		idx, err := table.Index("foo")
-		assert.Nil(t, err)
-		info, err := mapper.NewInfo(schema.Table("Open_vSwitch"), obj)
-		assert.Nil(t, err)
-		v, err := valueFromIndex(info, newIndex("foo"))
-		assert.Nil(t, err)
-		assert.Equal(t, idx[v], obj.UUID)
-	})
-	t.Run("Index by single column miss", func(t *testing.T) {
-		idx, err := table.Index("foo")
-		assert.Nil(t, err)
-		obj2 := obj
-		obj2.Foo = "wrong"
-		assert.Nil(t, err)
-		info, err := mapper.NewInfo(schema.Table("Open_vSwitch"), obj2)
-		assert.Nil(t, err)
-		v, err := valueFromIndex(info, newIndex("foo"))
-		assert.Nil(t, err)
-		_, ok := idx[v]
-		assert.False(t, ok)
-	})
-	t.Run("Index by single column wrong", func(t *testing.T) {
-		_, err := table.Index("wrong")
-		assert.NotNil(t, err)
-	})
-	t.Run("Index by multi-column wrong", func(t *testing.T) {
-		_, err := table.Index("bar", "wrong")
-		assert.NotNil(t, err)
-	})
-	t.Run("Index by multi-column", func(t *testing.T) {
-		idx, err := table.Index("bar", "baz")
-		assert.Nil(t, err)
-		info, err := mapper.NewInfo(schema.Table("Open_vSwitch"), obj)
-		assert.Nil(t, err)
-		v, err := valueFromIndex(info, newIndex("bar", "baz"))
-		assert.Nil(t, err)
-		assert.Equal(t, idx[v], obj.UUID)
-	})
-	t.Run("Index by multi-column miss", func(t *testing.T) {
-		idx, err := table.Index("bar", "baz")
-		assert.Nil(t, err)
-		obj2 := obj
-		obj2.Baz++
-		info, err := mapper.NewInfo(schema.Table("Open_vSwitch"), obj)
-		assert.Nil(t, err)
-		v, err := valueFromIndex(info, newIndex("bar", "baz"))
-		assert.Nil(t, err)
-		_, ok := idx[v]
-		assert.False(t, ok)
-	})
-	t.Run("Index type", func(t *testing.T) {
-		idx := newIndex("foo", "bar")
-		assert.Equal(t, idx.columns(), []string{"foo", "bar"})
-	})
-}
diff --git a/client/client_test.go b/client/client_test.go
index da09ee8..db24c81 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -4,11 +4,13 @@ import (
 	"encoding/json"
 	"reflect"
 	"strings"
-	"sync"
 	"testing"
 
+	"github.com/ovn-org/libovsdb/cache"
+	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 var (
@@ -18,6 +20,454 @@ var (
 	aUUID3 = "2f77b348-9768-4866-b761-89d5177ecda3"
 )
 
+type (
+	BridgeFailMode  = string
+	BridgeProtocols = string
+)
+
+const (
+	BridgeFailModeStandalone  BridgeFailMode  = "standalone"
+	BridgeFailModeSecure      BridgeFailMode  = "secure"
+	BridgeProtocolsOpenflow10 BridgeProtocols = "OpenFlow10"
+	BridgeProtocolsOpenflow11 BridgeProtocols = "OpenFlow11"
+	BridgeProtocolsOpenflow12 BridgeProtocols = "OpenFlow12"
+	BridgeProtocolsOpenflow13 BridgeProtocols = "OpenFlow13"
+	BridgeProtocolsOpenflow14 BridgeProtocols = "OpenFlow14"
+	BridgeProtocolsOpenflow15 BridgeProtocols = "OpenFlow15"
+)
+
+// Bridge defines an object in Bridge table
+type Bridge struct {
+	UUID                string            `ovsdb:"_uuid"`
+	AutoAttach          []string          `ovsdb:"auto_attach"`
+	Controller          []string          `ovsdb:"controller"`
+	DatapathID          []string          `ovsdb:"datapath_id"`
+	DatapathType        string            `ovsdb:"datapath_type"`
+	DatapathVersion     string            `ovsdb:"datapath_version"`
+	ExternalIDs         map[string]string `ovsdb:"external_ids"`
+	FailMode            []BridgeFailMode  `ovsdb:"fail_mode"`
+	FloodVLANs          []int             `ovsdb:"flood_vlans"`
+	FlowTables          map[int]string    `ovsdb:"flow_tables"`
+	IPFIX               []string          `ovsdb:"ipfix"`
+	McastSnoopingEnable bool              `ovsdb:"mcast_snooping_enable"`
+	Mirrors             []string          `ovsdb:"mirrors"`
+	Name                string            `ovsdb:"name"`
+	Netflow             []string          `ovsdb:"netflow"`
+	OtherConfig         map[string]string `ovsdb:"other_config"`
+	Ports               []string          `ovsdb:"ports"`
+	Protocols           []BridgeProtocols `ovsdb:"protocols"`
+	RSTPEnable          bool              `ovsdb:"rstp_enable"`
+	RSTPStatus          map[string]string `ovsdb:"rstp_status"`
+	Sflow               []string          `ovsdb:"sflow"`
+	Status              map[string]string `ovsdb:"status"`
+	STPEnable           bool              `ovsdb:"stp_enable"`
+}
+
+// OpenvSwitch defines an object in Open_vSwitch table
+type OpenvSwitch struct {
+	UUID            string            `ovsdb:"_uuid"`
+	Bridges         []string          `ovsdb:"bridges"`
+	CurCfg          int               `ovsdb:"cur_cfg"`
+	DatapathTypes   []string          `ovsdb:"datapath_types"`
+	Datapaths       map[string]string `ovsdb:"datapaths"`
+	DbVersion       []string          `ovsdb:"db_version"`
+	DpdkInitialized bool              `ovsdb:"dpdk_initialized"`
+	DpdkVersion     []string          `ovsdb:"dpdk_version"`
+	ExternalIDs     map[string]string `ovsdb:"external_ids"`
+	IfaceTypes      []string          `ovsdb:"iface_types"`
+	ManagerOptions  []string          `ovsdb:"manager_options"`
+	NextCfg         int               `ovsdb:"next_cfg"`
+	OtherConfig     map[string]string `ovsdb:"other_config"`
+	OVSVersion      []string          `ovsdb:"ovs_version"`
+	SSL             []string          `ovsdb:"ssl"`
+	Statistics      map[string]string `ovsdb:"statistics"`
+	SystemType      []string          `ovsdb:"system_type"`
+	SystemVersion   []string          `ovsdb:"system_version"`
+}
+
+var schema = `{
+	"name": "Open_vSwitch",
+	"version": "8.2.0",
+	"tables": {
+	  "Bridge": {
+		"columns": {
+		  "auto_attach": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "AutoAttach"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "controller": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "Controller"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "datapath_id": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": 1
+			},
+			"ephemeral": true
+		  },
+		  "datapath_type": {
+			"type": "string"
+		  },
+		  "datapath_version": {
+			"type": "string"
+		  },
+		  "external_ids": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "value": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "fail_mode": {
+			"type": {
+			  "key": {
+				"type": "string",
+				"enum": [
+				  "set",
+				  [
+					"standalone",
+					"secure"
+				  ]
+				]
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "flood_vlans": {
+			"type": {
+			  "key": {
+				"type": "integer",
+				"minInteger": 0,
+				"maxInteger": 4095
+			  },
+			  "min": 0,
+			  "max": 4096
+			}
+		  },
+		  "flow_tables": {
+			"type": {
+			  "key": {
+				"type": "integer",
+				"minInteger": 0,
+				"maxInteger": 254
+			  },
+			  "value": {
+				"type": "uuid",
+				"refTable": "Flow_Table"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "ipfix": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "IPFIX"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "mcast_snooping_enable": {
+			"type": "boolean"
+		  },
+		  "mirrors": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "Mirror"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "name": {
+			"type": "string",
+			"mutable": false
+		  },
+		  "netflow": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "NetFlow"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "other_config": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "value": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "ports": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "Port"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "protocols": {
+			"type": {
+			  "key": {
+				"type": "string",
+				"enum": [
+				  "set",
+				  [
+					"OpenFlow10",
+					"OpenFlow11",
+					"OpenFlow12",
+					"OpenFlow13",
+					"OpenFlow14",
+					"OpenFlow15"
+				  ]
+				]
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "rstp_enable": {
+			"type": "boolean"
+		  },
+		  "rstp_status": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "value": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			},
+			"ephemeral": true
+		  },
+		  "sflow": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "sFlow"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "status": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "value": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			},
+			"ephemeral": true
+		  },
+		  "stp_enable": {
+			"type": "boolean"
+		  }
+		},
+		"indexes": [
+		  [
+			"name"
+		  ]
+		]
+	  },
+	  "Open_vSwitch": {
+		"columns": {
+		  "bridges": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "Bridge"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "cur_cfg": {
+			"type": "integer"
+		  },
+		  "datapath_types": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "datapaths": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "value": {
+				"type": "uuid",
+				"refTable": "Datapath"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "db_version": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "dpdk_initialized": {
+			"type": "boolean"
+		  },
+		  "dpdk_version": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "external_ids": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "value": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "iface_types": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "manager_options": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "Manager"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "next_cfg": {
+			"type": "integer"
+		  },
+		  "other_config": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "value": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			}
+		  },
+		  "ovs_version": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "ssl": {
+			"type": {
+			  "key": {
+				"type": "uuid",
+				"refTable": "SSL"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "statistics": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "value": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": "unlimited"
+			},
+			"ephemeral": true
+		  },
+		  "system_type": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  },
+		  "system_version": {
+			"type": {
+			  "key": {
+				"type": "string"
+			  },
+			  "min": 0,
+			  "max": 1
+			}
+		  }
+		}
+	  }
+	}
+  }`
+
 func testOvsSet(t *testing.T, set interface{}) ovsdb.OvsSet {
 	oSet, err := ovsdb.NewOvsSet(set)
 	assert.Nil(t, err)
@@ -30,11 +480,7 @@ func testOvsMap(t *testing.T, set interface{}) ovsdb.OvsMap {
 	return oMap
 }
 
-func updateBenchmark(updates []byte, b *testing.B) {
-	ovs := OvsdbClient{
-		handlers:      []ovsdb.NotificationHandler{},
-		handlersMutex: &sync.Mutex{},
-	}
+func updateBenchmark(ovs *ovsdbClient, updates []byte, b *testing.B) {
 	for n := 0; n < b.N; n++ {
 		params := []json.RawMessage{[]byte(`"v1"`), updates}
 		var reply []interface{}
@@ -73,8 +519,12 @@ func newBridgeRow(name string) string {
 }
 
 func newOvsRow(bridges ...string) string {
+	bridgeUUIDs := []string{}
+	for _, b := range bridges {
+		bridgeUUIDs = append(bridgeUUIDs, `[ "uuid", "`+b+`" ]`)
+	}
 	return `{
-		"bridges": [ "set", ["` + strings.Join(bridges, `","`) + `"]],
+		"bridges": [ "set", [` + strings.Join(bridgeUUIDs, `,`) + `]],
 		"cur_cfg": 0,
 		"datapath_types": [ "set", [] ],
 		"datapaths": [ "map", [] ],
@@ -83,7 +533,7 @@ func newOvsRow(bridges ...string) string {
 		"dpdk_version":     [ "set", [] ],
 		"external_ids":     [ "map", [["system-id","829f8534-94a8-468e-9176-132738cf260a"]]],
 		"iface_types":      [ "set", [] ],
-		"manager_options":  "6e4cd5fc-f51a-462a-b3d6-a696af6d7a84",
+		"manager_options":  ["uuid", "6e4cd5fc-f51a-462a-b3d6-a696af6d7a84"],
 		"next_cfg":         0,
 		"other_config":     [ "map", [] ],
 		"ovs_version":      "2.15.90",
@@ -95,87 +545,145 @@ func newOvsRow(bridges ...string) string {
 }
 
 func BenchmarkUpdate1(b *testing.B) {
+	ovs, err := newOVSDBClient(defDB)
+	require.NoError(b, err)
+	var s ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(schema), &s)
+	require.NoError(b, err)
+	dbModel, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{
+		"Bridge":       &Bridge{},
+		"Open_vSwitch": &OpenvSwitch{},
+	})
+	require.NoError(b, err)
+	ovs.cache, err = cache.NewTableCache(&s, dbModel, nil)
+	require.NoError(b, err)
 	update := []byte(`{
 		"Open_vSwitch": {
-			"ovs": ` + newOvsRow("foo") + `
+			"ovs": {"new": ` + newOvsRow("foo") + `}
 		},
 		"Bridge": {
-			"foo": ` + newBridgeRow("foo") + `
+			"foo": {"new": ` + newBridgeRow("foo") + `}
 		}
 	}`)
-	updateBenchmark(update, b)
+	updateBenchmark(ovs, update, b)
 }
 
 func BenchmarkUpdate2(b *testing.B) {
+	ovs, err := newOVSDBClient(defDB)
+	require.NoError(b, err)
+	var s ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(schema), &s)
+	require.NoError(b, err)
+	dbModel, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{
+		"Bridge":       &Bridge{},
+		"Open_vSwitch": &OpenvSwitch{},
+	})
+	require.NoError(b, err)
+	ovs.cache, err = cache.NewTableCache(&s, dbModel, nil)
+	require.NoError(b, err)
 	update := []byte(`{
 		"Open_vSwitch": {
-			"ovs": ` + newOvsRow("foo", "bar") + `
+			"ovs": {"new": ` + newOvsRow("foo", "bar") + `}
 		},
 		"Bridge": {
-			"foo": ` + newBridgeRow("foo") + `,
-			"bar": ` + newBridgeRow("bar") + `
+			"foo": {"new": ` + newBridgeRow("foo") + `},
+			"bar": {"new": ` + newBridgeRow("bar") + `}
 		}
 	}`)
-	updateBenchmark(update, b)
+	updateBenchmark(ovs, update, b)
 }
 
 func BenchmarkUpdate3(b *testing.B) {
+	ovs, err := newOVSDBClient(defDB)
+	require.NoError(b, err)
+	var s ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(schema), &s)
+	require.NoError(b, err)
+	dbModel, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{
+		"Bridge":       &Bridge{},
+		"Open_vSwitch": &OpenvSwitch{},
+	})
+	require.NoError(b, err)
+	ovs.cache, err = cache.NewTableCache(&s, dbModel, nil)
+	require.NoError(b, err)
 	update := []byte(`{
 		"Open_vSwitch": {
-			"ovs": ` + newOvsRow("foo", "bar", "baz") + `
+			"ovs": {"new": ` + newOvsRow("foo", "bar", "baz") + `}
 		},
 		"Bridge": {
-			"foo": ` + newBridgeRow("foo") + `,
-			"bar": ` + newBridgeRow("bar") + `,
-			"baz": ` + newBridgeRow("baz") + `
+			"foo": {"new": ` + newBridgeRow("foo") + `},
+			"bar": {"new": ` + newBridgeRow("bar") + `},
+			"baz": {"new": ` + newBridgeRow("baz") + `}
 		}
 	}`)
-	updateBenchmark(update, b)
+	updateBenchmark(ovs, update, b)
 }
 
 func BenchmarkUpdate5(b *testing.B) {
+	ovs, err := newOVSDBClient(defDB)
+	require.NoError(b, err)
+	var s ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(schema), &s)
+	require.NoError(b, err)
+	dbModel, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{
+		"Bridge":       &Bridge{},
+		"Open_vSwitch": &OpenvSwitch{},
+	})
+	require.NoError(b, err)
+	ovs.cache, err = cache.NewTableCache(&s, dbModel, nil)
+	require.NoError(b, err)
 	update := []byte(`{
 		"Open_vSwitch": {
-			"ovs": ` + newOvsRow("foo", "bar", "baz", "quux", "foofoo") + `
+			"ovs": {"new": ` + newOvsRow("foo", "bar", "baz", "quux", "foofoo") + `}
 		},
 		"Bridge": {
-			"foo": ` + newBridgeRow("foo") + `,
-			"bar": ` + newBridgeRow("bar") + `,
-			"baz": ` + newBridgeRow("baz") + `,
-			"quux": ` + newBridgeRow("quux") + `,
-			"foofoo": ` + newBridgeRow("foofoo") + `
+			"foo": {"new": ` + newBridgeRow("foo") + `},
+			"bar": {"new": ` + newBridgeRow("bar") + `},
+			"baz": {"new": ` + newBridgeRow("baz") + `},
+			"quux": {"new": ` + newBridgeRow("quux") + `},
+			"foofoo": {"new": ` + newBridgeRow("foofoo") + `}
 		}
 	}`)
-	updateBenchmark(update, b)
+	updateBenchmark(ovs, update, b)
 }
 
 func BenchmarkUpdate8(b *testing.B) {
+	ovs, err := newOVSDBClient(defDB)
+	require.NoError(b, err)
+	var s ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(schema), &s)
+	require.NoError(b, err)
+	dbModel, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{
+		"Bridge":       &Bridge{},
+		"Open_vSwitch": &OpenvSwitch{},
+	})
+	require.NoError(b, err)
+	ovs.cache, err = cache.NewTableCache(&s, dbModel, nil)
+	require.NoError(b, err)
 	update := []byte(`{
 		"Open_vSwitch": {
-			"ovs": ` + newOvsRow("foo", "bar", "baz", "quux", "foofoo", "foobar", "foobaz", "fooquux") + `
+			"ovs": {"new": ` + newOvsRow("foo", "bar", "baz", "quux", "foofoo", "foobar", "foobaz", "fooquux") + `}
 		},
 		"Bridge": {
-			"foo": ` + newBridgeRow("foo") + `,
-			"bar": ` + newBridgeRow("bar") + `,
-			"baz": ` + newBridgeRow("baz") + `,
-			"quux": ` + newBridgeRow("quux") + `,
-			"foofoo": ` + newBridgeRow("foofoo") + `,
-			"foobar": ` + newBridgeRow("foobar") + `,
-			"foobaz": ` + newBridgeRow("foobaz") + `,
-			"fooquux": ` + newBridgeRow("fooquux") + `
+			"foo": {"new": ` + newBridgeRow("foo") + `},
+			"bar": {"new": ` + newBridgeRow("bar") + `},
+			"baz": {"new": ` + newBridgeRow("baz") + `},
+			"quux": {"new": ` + newBridgeRow("quux") + `},
+			"foofoo": {"new": ` + newBridgeRow("foofoo") + `},
+			"foobar": {"new": ` + newBridgeRow("foobar") + `},
+			"foobaz": {"new": ` + newBridgeRow("foobaz") + `},
+			"fooquux": {"new": ` + newBridgeRow("fooquux") + `}
 		}
 	}`)
-	updateBenchmark(update, b)
+	updateBenchmark(ovs, update, b)
 }
 
 func TestEcho(t *testing.T) {
 	req := []interface{}{"hi"}
 	var reply []interface{}
-	ovs := OvsdbClient{
-		handlers:      []ovsdb.NotificationHandler{},
-		handlersMutex: &sync.Mutex{},
-	}
-	err := ovs.echo(req, &reply)
+	ovs, err := newOVSDBClient(defDB)
+	require.NoError(t, err)
+	err = ovs.echo(req, &reply)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -185,22 +693,91 @@ func TestEcho(t *testing.T) {
 }
 
 func TestUpdate(t *testing.T) {
-	ovs := OvsdbClient{
-		handlers:      []ovsdb.NotificationHandler{},
-		handlersMutex: &sync.Mutex{},
-	}
+	ovs, err := newOVSDBClient(defDB)
+	require.NoError(t, err)
+	var s ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(schema), &s)
+	require.NoError(t, err)
+	dbModel, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{
+		"Bridge":       &Bridge{},
+		"Open_vSwitch": &OpenvSwitch{},
+	})
+	require.NoError(t, err)
+	ovs.cache, err = cache.NewTableCache(&s, dbModel, nil)
+	require.NoError(t, err)
 	var reply []interface{}
-	validUpdate := ovsdb.TableUpdates{
-		"table": {
-			"uuid": &ovsdb.RowUpdate{},
+	update := []byte(`{
+		"Open_vSwitch": {
+			"ovs": {"new": ` + newOvsRow("foo") + `}
 		},
-	}
-	b, err := json.Marshal(validUpdate)
-	if err != nil {
-		t.Fatal(err)
-	}
-	err = ovs.update([]json.RawMessage{[]byte(`"hello"`), b}, &reply)
+		"Bridge": {
+			"foo": {"new": ` + newBridgeRow("foo") + `}
+		}
+	}`)
+	params := []json.RawMessage{[]byte(`"v1"`), update}
+	err = ovs.update(params, &reply)
 	if err != nil {
 		t.Error(err)
 	}
 }
+
+func TestOperationWhenNotConnected(t *testing.T) {
+	ovs, err := newOVSDBClient(defDB)
+	require.NoError(t, err)
+	tests := []struct {
+		name string
+		fn   func() error
+	}{
+		{
+			"echo",
+			func() error {
+				return ovs.Echo()
+			},
+		},
+		{
+			"transact",
+			func() error {
+				_, err := ovs.Transact(ovsdb.Operation{})
+				return err
+			},
+		},
+		{
+			"monitor",
+			func() error {
+				return ovs.Monitor("")
+			},
+		},
+		{
+			"monitor all",
+			func() error {
+				return ovs.MonitorAll("")
+			},
+		},
+		{
+			"monitor cancel",
+			func() error {
+				return ovs.MonitorCancel("")
+			},
+		},
+	}
+	for _, tt := range tests {
+		err := tt.fn()
+		assert.EqualError(t, err, ErrNotConnected.Error())
+	}
+}
+
+func TestSetOption(t *testing.T) {
+	o, err := newOVSDBClient(defDB)
+	require.NoError(t, err)
+
+	o.options, err = newOptions()
+	require.NoError(t, err)
+
+	err = o.SetOption(WithEndpoint("tcp::6640"))
+	require.NoError(t, err)
+
+	o.connected = true
+
+	err = o.SetOption(WithEndpoint("tcp::6641"))
+	assert.EqualError(t, err, "cannot set option when client is connected")
+}
diff --git a/client/ovs_integration_test.go b/client/ovs_integration_test.go
index c6ee311..db6dcff 100644
--- a/client/ovs_integration_test.go
+++ b/client/ovs_integration_test.go
@@ -11,6 +11,7 @@ import (
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 const (
@@ -62,26 +63,21 @@ func TestConnectIntegration(t *testing.T) {
 	if testing.Short() {
 		t.Skip()
 	}
-	timeoutChan := make(chan bool)
 	connected := make(chan bool)
-	go func() {
-		time.Sleep(10 * time.Second)
-		timeoutChan <- true
-	}()
+	errs := make(chan error)
 
-	go func() {
-		// Use Convenience params. Ignore failure even if any
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
 
-		_, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
+	go func() {
+		ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
 		if err != nil {
-			log.Println("Couldnt establish OVSDB connection with Defult params. No big deal")
+			errs <- err
+			return
 		}
-	}()
-
-	go func() {
-		ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
+		err = ovs.Connect(ctx)
 		if err != nil {
-			connected <- false
+			errs <- err
 		} else {
 			connected <- true
 			ovs.Disconnect()
@@ -89,16 +85,16 @@ func TestConnectIntegration(t *testing.T) {
 	}()
 
 	select {
-	case <-timeoutChan:
-		t.Error("Connection Timed Out")
-	case b := <-connected:
-		if !b {
-			t.Error("Couldnt connect to OVSDB Server")
-		}
+	case err := <-errs:
+		t.Fatal(err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("Connection Timed Out")
+	case <-connected:
+		return
 	}
 }
 
-func TestListDbsIntegration(t *testing.T) {
+func TestConnectReconnectIntegration(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
@@ -107,12 +103,46 @@ func TestListDbsIntegration(t *testing.T) {
 		t.Skip()
 	}
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
-	reply, err := ovs.ListDbs()
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
 
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(ctx)
+	require.NoError(t, err)
+
+	err = ovs.Echo()
+	require.NoError(t, err)
+
+	ovs.Disconnect()
+
+	err = ovs.Echo()
+	require.EqualError(t, err, ErrNotConnected.Error())
+
+	ctx, cancel = context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	err = ovs.Connect(ctx)
+	require.NoError(t, err)
+
+	err = ovs.Echo()
+	assert.NoError(t, err)
+
+}
+
+func TestListDbsIntegration(t *testing.T) {
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+	SetConfig()
+	if testing.Short() {
+		t.Skip()
+	}
+	ovs, err := newOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
+	reply, err := ovs.listDbs()
 	if err != nil {
 		log.Fatal("ListDbs error:", err)
 	}
@@ -120,7 +150,6 @@ func TestListDbsIntegration(t *testing.T) {
 	found := false
 	for _, db := range reply {
 		if db == "Open_vSwitch" {
-			log.Println("Couldnt establish OVSDB connection with Defult params. No big deal")
 			found = true
 		}
 	}
@@ -129,7 +158,7 @@ func TestListDbsIntegration(t *testing.T) {
 		t.Error("Expected: 'Open_vSwitch'", reply)
 	}
 	var b bytes.Buffer
-	ovs.Schema.Print(&b)
+	ovs.Schema().Print(&b)
 	ovs.Disconnect()
 }
 
@@ -142,13 +171,12 @@ func TestGetSchemasIntegration(t *testing.T) {
 		t.Skip()
 	}
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
-
+	ovs, err := newOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 	dbName := "Open_vSwitch"
-	reply, err := ovs.GetSchema(dbName)
+	reply, err := ovs.getSchema(dbName)
 
 	if err != nil {
 		log.Fatal("GetSchemas error:", err)
@@ -170,10 +198,10 @@ func TestInsertTransactIntegration(t *testing.T) {
 	}
 	SetConfig()
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 	err = ovs.MonitorAll(nil)
 	assert.Nil(t, err)
 
@@ -228,10 +256,10 @@ func TestDeleteTransactIntegration(t *testing.T) {
 		t.Skip()
 	}
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 	err = ovs.MonitorAll(nil)
 	assert.Nil(t, err)
 
@@ -273,11 +301,10 @@ func TestMonitorIntegration(t *testing.T) {
 		t.Skip()
 	}
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
-
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 	err = ovs.MonitorAll(nil)
 	if err != nil {
 		t.Fatal(err)
@@ -285,90 +312,16 @@ func TestMonitorIntegration(t *testing.T) {
 	ovs.Disconnect()
 }
 
-func TestNotifyIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
-
-	notifyEchoChan := make(chan bool)
-
-	notifier := Notifier{notifyEchoChan}
-	ovs.Register(notifier)
-
-	timeoutChan := make(chan bool)
-	go func() {
-		time.Sleep(10 * time.Second)
-		timeoutChan <- true
-	}()
-
-	select {
-	case <-timeoutChan:
-		t.Fatal("timed out")
-	case <-notifyEchoChan:
-		break
-	}
-	ovs.Disconnect()
-}
-
-func TestRemoveNotifyIntegration(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping integration test")
-	}
-	SetConfig()
-
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
-
-	notifyEchoChan := make(chan bool)
-
-	notifier := Notifier{notifyEchoChan}
-	ovs.Register(notifier)
-
-	lenIni := len(ovs.handlers)
-	_ = ovs.Unregister(notifier)
-	lenEnd := len(ovs.handlers)
-
-	if lenIni == lenEnd {
-		log.Fatal("Failed to Unregister Notifier:")
-	}
-
-	ovs.Disconnect()
-}
-
-type Notifier struct {
-	echoChan chan bool
-}
-
-func (n Notifier) Update(interface{}, ovsdb.TableUpdates) {
-}
-func (n Notifier) Locked([]interface{}) {
-}
-func (n Notifier) Stolen([]interface{}) {
-}
-func (n Notifier) Echo([]interface{}) {
-	n.echoChan <- true
-}
-func (n Notifier) Disconnected() {
-}
-
 func TestTableSchemaValidationIntegration(t *testing.T) {
 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}
 	SetConfig()
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 
 	operation := ovsdb.Operation{
 		Op:    "insert",
@@ -390,10 +343,10 @@ func TestColumnSchemaInRowValidationIntegration(t *testing.T) {
 	}
 	SetConfig()
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 
 	operation := ovsdb.Operation{
 		Op:    "insert",
@@ -416,10 +369,10 @@ func TestColumnSchemaInMultipleRowsValidationIntegration(t *testing.T) {
 	}
 	SetConfig()
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 
 	invalidBridge := ovsdb.Row(map[string]interface{}{"invalid_column": "invalid_column"})
 	bridge := ovsdb.Row(map[string]interface{}{"name": "docker-ovs"})
@@ -448,10 +401,10 @@ func TestColumnSchemaValidationIntegration(t *testing.T) {
 		t.Skip()
 	}
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 
 	operation := ovsdb.Operation{
 		Op:      "select",
@@ -473,10 +426,10 @@ func TestMonitorCancelIntegration(t *testing.T) {
 	}
 	SetConfig()
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 
 	monitorID := "f1b2ca48-aad7-11e7-abc4-cec278b6b50a"
 
@@ -508,10 +461,11 @@ func TestInsertDuplicateTransactIntegration(t *testing.T) {
 	}
 	SetConfig()
 
-	ovs, err := Connect(context.Background(), defDB, WithEndpoint(cfg.Addr))
-	if err != nil {
-		t.Fatalf("Failed to Connect. error: %s", err)
-	}
+	ovs, err := NewOVSDBClient(defDB, WithEndpoint(cfg.Addr))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
+
 	err = ovs.MonitorAll(nil)
 	assert.Nil(t, err)
 
diff --git a/ovsdb/bindings_test.go b/ovsdb/bindings_test.go
index 50c90f7..6d8bd3a 100644
--- a/ovsdb/bindings_test.go
+++ b/ovsdb/bindings_test.go
@@ -48,12 +48,6 @@ var (
 		"key3": "value3",
 	}
 
-	aUUIDMap = map[string]string{
-		"key1": aUUID0,
-		"key2": aUUID1,
-		"key3": aUUID2,
-	}
-
 	aEmptySet = []string{}
 )
 
@@ -81,12 +75,6 @@ func TestOvsToNativeAndNativeToOvs(t *testing.T) {
 
 	m, _ := NewOvsMap(aMap)
 
-	um, _ := NewOvsMap(map[string]UUID{
-		"key1": {GoUUID: aUUID0},
-		"key2": {GoUUID: aUUID1},
-		"key3": {GoUUID: aUUID2},
-	})
-
 	tests := []struct {
 		name   string
 		schema []byte
@@ -342,20 +330,6 @@ func TestOvsToNativeAndNativeToOvs(t *testing.T) {
 			native: aMap,
 			ovs:    m,
 		},
-		{
-			name: "Map (string->uuid)",
-			schema: []byte(`{
-			"type": {
-				"key": "string",
-				"max": "unlimited",
-				"min": 0,
-				"value": "uuid"
-			}
-			}`),
-			input:  um,
-			native: aUUIDMap,
-			ovs:    um,
-		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
diff --git a/server/server_integration_test.go b/server/server_integration_test.go
index 7fdec6d..0bfdc8a 100644
--- a/server/server_integration_test.go
+++ b/server/server_integration_test.go
@@ -82,9 +82,10 @@ func TestClientServerEcho(t *testing.T) {
 		return server.Ready()
 	}, 1*time.Second, 10*time.Millisecond)
 
-	ovs, err := client.Connect(context.Background(), defDB, client.WithEndpoint(fmt.Sprintf("unix:%s", tmpfile)))
-	require.Nil(t, err)
-
+	ovs, err := client.NewOVSDBClient(defDB, client.WithEndpoint(fmt.Sprintf("unix:%s", tmpfile)))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 	err = ovs.Echo()
 	assert.Nil(t, err)
 }
@@ -118,8 +119,10 @@ func TestClientServerInsert(t *testing.T) {
 		return server.Ready()
 	}, 1*time.Second, 10*time.Millisecond)
 
-	ovs, err := client.Connect(context.Background(), defDB, client.WithEndpoint(fmt.Sprintf("unix:%s", tmpfile)))
-	require.Nil(t, err)
+	ovs, err := client.NewOVSDBClient(defDB, client.WithEndpoint(fmt.Sprintf("unix:%s", tmpfile)))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 
 	bridgeRow := &bridgeType{
 		Name:        "foo",
@@ -167,8 +170,10 @@ func TestClientServerMonitor(t *testing.T) {
 		return server.Ready()
 	}, 1*time.Second, 10*time.Millisecond)
 
-	ovs, err := client.Connect(context.Background(), defDB, client.WithEndpoint(fmt.Sprintf("unix:%s", tmpfile)))
-	require.Nil(t, err)
+	ovs, err := client.NewOVSDBClient(defDB, client.WithEndpoint(fmt.Sprintf("unix:%s", tmpfile)))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 
 	ovsRow := &ovsType{
 		UUID: "ovs",
@@ -183,7 +188,7 @@ func TestClientServerMonitor(t *testing.T) {
 	seenInsert := false
 	seenMutation := false
 	seenInitialOvs := false
-	ovs.Cache.AddEventHandler(&cache.EventHandlerFuncs{
+	ovs.Cache().AddEventHandler(&cache.EventHandlerFuncs{
 		AddFunc: func(table string, model model.Model) {
 			if table == "Bridge" {
 				br := model.(*bridgeType)
@@ -289,8 +294,10 @@ func TestClientServerInsertAndDelete(t *testing.T) {
 		return server.Ready()
 	}, 1*time.Second, 10*time.Millisecond)
 
-	ovs, err := client.Connect(context.Background(), defDB, client.WithEndpoint(fmt.Sprintf("unix:%s", tmpfile)))
-	require.Nil(t, err)
+	ovs, err := client.NewOVSDBClient(defDB, client.WithEndpoint(fmt.Sprintf("unix:%s", tmpfile)))
+	require.NoError(t, err)
+	err = ovs.Connect(context.Background())
+	require.NoError(t, err)
 
 	bridgeRow := &bridgeType{
 		Name:        "foo",