mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

DB driver implementation via RPC (#17779)

Просмотр исходного кода 
This PR builds up on the pass-through DB driver to a fully functioning DB driver implementation via our RPC layer.

To keep things separate from the plugin RPC API, and have the ability to move fast with changes, a separate field Driver is added to MattermostPlugin. Typically the field which is required to be compatible are the API and Helpers. It would be well-documented that Driver is purely for internal use by Mattermost plugins.

A new Driver interface was created which would have a client and server implementation. Every object (connection, statement, etc.) is created and added to a map on the server side. On the client side, the wrapper structs hold the object id, and communicate via the RPC API using this id.

When the server gets the object id, it picks up the appropriate object from its map and performs the operation, and sends back the data.

Some things that need to be handled are errors. Typical error types like pq.Error and mysql.MySQLError are registered with encoding/gob. But for error variables like sql.ErrNoRows, a special integer is encoded with the ErrorString struct. And on the cilent side, the integer is checked, and the appropriate error variable is returned.

Some pending things:

- Context support. This is tricky. Since context.Context is an interface, it's not possible to marshal it. We have to find a way to get the timeout value from the context and pass it.
- RowsColumnScanType(rowsID string, index int) reflect.Type API. Again, reflect.Type is an interface.
- Master/Replica API support.
Этот коммит содержится в:
Agniva De Sarker
2021-06-17 08:53:52 +05:30
коммит произвёл GitHub
родитель f0bd973a5c
Коммит 4b95d47923
20 изменённых файлов: 1176 добавлений и 174 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

7
plugin/client.go
Просмотреть файл

@@ -19,11 +19,13 @@ func ClientMain(pluginImplementation interface{}) {
if impl, ok := pluginImplementation.(interface {
SetAPI(api API)
SetHelpers(helpers Helpers)
SetDriver(driver Driver)
}); !ok {
panic("Plugin implementation given must embed plugin.MattermostPlugin")
} else {
impl.SetAPI(nil)
impl.SetHelpers(nil)
impl.SetDriver(nil)
}
pluginMap := map[string]plugin.Plugin{
@@ -40,6 +42,7 @@ type MattermostPlugin struct {
// API exposes the plugin api, and becomes available just prior to the OnActive hook.
API API
Helpers Helpers
Driver Driver
}
// SetAPI persists the given API interface to the plugin. It is invoked just prior to the
@@ -52,3 +55,7 @@ func (p *MattermostPlugin) SetAPI(api API) {
func (p *MattermostPlugin) SetHelpers(helpers Helpers) {
p.Helpers = helpers
}
func (p *MattermostPlugin) SetDriver(driver Driver) {
p.Driver = driver
}

97
plugin/client_rpc.go
Просмотреть файл

@@ -7,6 +7,8 @@ package plugin
import (
"bytes"
"database/sql"
"database/sql/driver"
"encoding/gob"
"encoding/json"
"fmt"
@@ -19,7 +21,9 @@ import (
"reflect"
"github.com/dyatlov/go-opengraph/opengraph"
"github.com/go-sql-driver/mysql"
"github.com/hashicorp/go-plugin"
"github.com/lib/pq"
"github.com/mattermost/mattermost-server/v5/model"
"github.com/mattermost/mattermost-server/v5/shared/mlog"
@@ -32,6 +36,7 @@ type hooksRPCClient struct {
log *mlog.Logger
muxBroker *plugin.MuxBroker
apiImpl API
driver Driver
implemented [TotalHooksID]bool
}
@@ -43,9 +48,10 @@ type hooksRPCServer struct {
// Implements hashicorp/go-plugin/plugin.Plugin interface to connect the hooks of a plugin
type hooksPlugin struct {
hooks interface{}
apiImpl API
log *mlog.Logger
hooks interface{}
apiImpl API
driverImpl Driver
log *mlog.Logger
}
func (p *hooksPlugin) Server(b *plugin.MuxBroker) (interface{}, error) {
@@ -53,7 +59,12 @@ func (p *hooksPlugin) Server(b *plugin.MuxBroker) (interface{}, error) {
}
func (p *hooksPlugin) Client(b *plugin.MuxBroker, client *rpc.Client) (interface{}, error) {
return &hooksRPCClient{client: client, log: p.log, muxBroker: b, apiImpl: p.apiImpl}, nil
return &hooksRPCClient{client: client,
log: p.log,
muxBroker: b,
apiImpl: p.apiImpl,
driver: p.driverImpl,
}, nil
}
type apiRPCClient struct {
@@ -72,7 +83,8 @@ type apiRPCServer struct {
// ErrorString merely preserves the string description of the error, while satisfying the error
// interface itself to allow other registered types (such as model.AppError) to be sent unmodified.
type ErrorString struct {
Err string
Code int // Code to map to various error variables
Err string
}
func (e ErrorString) Error() string {
@@ -87,9 +99,58 @@ func encodableError(err error) error {
return err
}
return &ErrorString{
if _, ok := err.(*pq.Error); ok {
return err
}
if _, ok := err.(*mysql.MySQLError); ok {
return err
}
ret := &ErrorString{
Err: err.Error(),
}
switch err {
case io.EOF:
ret.Code = 1
case sql.ErrNoRows:
ret.Code = 2
case sql.ErrConnDone:
ret.Code = 3
case sql.ErrTxDone:
ret.Code = 4
case driver.ErrSkip:
ret.Code = 5
case driver.ErrBadConn:
ret.Code = 6
case driver.ErrRemoveArgument:
ret.Code = 7
}
return ret
}
func decodableError(err error) error {
if encErr, ok := err.(*ErrorString); ok {
switch encErr.Code {
case 1:
return io.EOF
case 2:
return sql.ErrNoRows
case 3:
return sql.ErrConnDone
case 4:
return sql.ErrTxDone
case 5:
return driver.ErrSkip
case 6:
return driver.ErrBadConn
case 7:
return driver.ErrRemoveArgument
}
}
return err
}
// Registering some types used by MM for encoding/gob used by rpc
@@ -98,6 +159,8 @@ func init() {
gob.Register([]interface{}{})
gob.Register(map[string]interface{}{})
gob.Register(&model.AppError{})
gob.Register(&pq.Error{})
gob.Register(&mysql.MySQLError{})
gob.Register(&ErrorString{})
gob.Register(&opengraph.OpenGraph{})
gob.Register(&model.AutocompleteDynamicListArg{})
@@ -167,7 +230,8 @@ func (s *hooksRPCServer) Implemented(args struct{}, reply *[]string) error {
}
type Z_OnActivateArgs struct {
APIMuxId uint32
APIMuxId uint32
DriverMuxId uint32
}
type Z_OnActivateReturns struct {
@@ -181,8 +245,14 @@ func (g *hooksRPCClient) OnActivate() error {
muxBroker: g.muxBroker,
})
nextID := g.muxBroker.NextId()
go g.muxBroker.AcceptAndServe(nextID, &dbRPCServer{
dbImpl: g.driver,
})
_args := &Z_OnActivateArgs{
APIMuxId: muxId,
APIMuxId: muxId,
DriverMuxId: nextID,
}
_returns := &Z_OnActivateReturns{}
@@ -198,17 +268,28 @@ func (s *hooksRPCServer) OnActivate(args *Z_OnActivateArgs, returns *Z_OnActivat
return err
}
conn2, err := s.muxBroker.Dial(args.DriverMuxId)
if err != nil {
return err
}
s.apiRPCClient = &apiRPCClient{
client: rpc.NewClient(connection),
muxBroker: s.muxBroker,
}
dbClient := &dbRPCClient{
client: rpc.NewClient(conn2),
}
if mmplugin, ok := s.impl.(interface {
SetAPI(api API)
SetHelpers(helpers Helpers)
SetDriver(driver Driver)
}); ok {
mmplugin.SetAPI(s.apiRPCClient)
mmplugin.SetHelpers(&HelpersImpl{API: s.apiRPCClient})
mmplugin.SetDriver(dbClient)
}
if mmplugin, ok := s.impl.(interface {

461
plugin/db_rpc.go Обычный файл
Просмотреть файл

@@ -0,0 +1,461 @@
// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
// See LICENSE.txt for license information.
package plugin
import (
"database/sql/driver"
"log"
"net/rpc"
)
// dbRPCClient contains the client-side logic to handle the RPC communication
// with the server. It's API is hand-written because we do not expect
// new methods to be added very frequently.
type dbRPCClient struct {
client *rpc.Client
}
// dbRPCServer is the server-side component which is responsible for calling
// the driver methods and properly encoding the responses back to the RPC client.
type dbRPCServer struct {
dbImpl Driver
}
var _ Driver = &dbRPCClient{}
type Z_DbStrErrReturn struct {
A string
B error
}
type Z_DbErrReturn struct {
A error
}
type Z_DbInt64ErrReturn struct {
A int64
B error
}
type Z_DbBoolReturn struct {
A bool
}
func (db *dbRPCClient) Conn() (string, error) {
ret := &Z_DbStrErrReturn{}
err := db.client.Call("Plugin.Conn", struct{}{}, ret)
if err != nil {
log.Printf("error during Plugin.Conn: %v", err)
}
ret.B = decodableError(ret.B)
return ret.A, ret.B
}
func (db *dbRPCServer) Conn(_ struct{}, ret *Z_DbStrErrReturn) error {
ret.A, ret.B = db.dbImpl.Conn()
ret.B = encodableError(ret.B)
return nil
}
func (db *dbRPCClient) ConnPing(connID string) error {
ret := &Z_DbErrReturn{}
err := db.client.Call("Plugin.ConnPing", connID, ret)
if err != nil {
log.Printf("error during Plugin.ConnPing: %v", err)
}
ret.A = decodableError(ret.A)
return ret.A
}
func (db *dbRPCServer) ConnPing(connID string, ret *Z_DbErrReturn) error {
ret.A = db.dbImpl.ConnPing(connID)
ret.A = encodableError(ret.A)
return nil
}
func (db *dbRPCClient) ConnClose(connID string) error {
ret := &Z_DbErrReturn{}
err := db.client.Call("Plugin.ConnClose", connID, ret)
if err != nil {
log.Printf("error during Plugin.ConnClose: %v", err)
}
ret.A = decodableError(ret.A)
return ret.A
}
func (db *dbRPCServer) ConnClose(connID string, ret *Z_DbErrReturn) error {
ret.A = db.dbImpl.ConnClose(connID)
ret.A = encodableError(ret.A)
return nil
}
type Z_DbTxArgs struct {
A string
B driver.TxOptions
}
func (db *dbRPCClient) Tx(connID string, opts driver.TxOptions) (string, error) {
args := &Z_DbTxArgs{
A: connID,
B: opts,
}
ret := &Z_DbStrErrReturn{}
err := db.client.Call("Plugin.Tx", args, ret)
if err != nil {
log.Printf("error during Plugin.Tx: %v", err)
}
ret.B = decodableError(ret.B)
return ret.A, ret.B
}
func (db *dbRPCServer) Tx(args *Z_DbTxArgs, ret *Z_DbStrErrReturn) error {
ret.A, ret.B = db.dbImpl.Tx(args.A, args.B)
ret.B = encodableError(ret.B)
return nil
}
func (db *dbRPCClient) TxCommit(txID string) error {
ret := &Z_DbErrReturn{}
err := db.client.Call("Plugin.TxCommit", txID, ret)
if err != nil {
log.Printf("error during Plugin.TxCommit: %v", err)
}
ret.A = decodableError(ret.A)
return ret.A
}
func (db *dbRPCServer) TxCommit(txID string, ret *Z_DbErrReturn) error {
ret.A = db.dbImpl.TxCommit(txID)
ret.A = encodableError(ret.A)
return nil
}
func (db *dbRPCClient) TxRollback(txID string) error {
ret := &Z_DbErrReturn{}
err := db.client.Call("Plugin.TxRollback", txID, ret)
if err != nil {
log.Printf("error during Plugin.TxRollback: %v", err)
}
ret.A = decodableError(ret.A)
return ret.A
}
func (db *dbRPCServer) TxRollback(txID string, ret *Z_DbErrReturn) error {
ret.A = db.dbImpl.TxRollback(txID)
ret.A = encodableError(ret.A)
return nil
}
type Z_DbStmtArgs struct {
A string
B string
}
func (db *dbRPCClient) Stmt(connID, q string) (string, error) {
args := &Z_DbStmtArgs{
A: connID,
B: q,
}
ret := &Z_DbStrErrReturn{}
err := db.client.Call("Plugin.Stmt", args, ret)
if err != nil {
log.Printf("error during Plugin.Stmt: %v", err)
}
ret.B = decodableError(ret.B)
return ret.A, ret.B
}
func (db *dbRPCServer) Stmt(args *Z_DbStmtArgs, ret *Z_DbStrErrReturn) error {
ret.A, ret.B = db.dbImpl.Stmt(args.A, args.B)
ret.B = encodableError(ret.B)
return nil
}
func (db *dbRPCClient) StmtClose(stID string) error {
ret := &Z_DbErrReturn{}
err := db.client.Call("Plugin.StmtClose", stID, ret)
if err != nil {
log.Printf("error during Plugin.StmtClose: %v", err)
}
ret.A = decodableError(ret.A)
return ret.A
}
func (db *dbRPCServer) StmtClose(stID string, ret *Z_DbErrReturn) error {
ret.A = db.dbImpl.StmtClose(stID)
ret.A = encodableError(ret.A)
return nil
}
type Z_DbIntReturn struct {
A int
}
func (db *dbRPCClient) StmtNumInput(stID string) int {
ret := &Z_DbIntReturn{}
err := db.client.Call("Plugin.StmtNumInput", stID, ret)
if err != nil {
log.Printf("error during Plugin.StmtNumInput: %v", err)
}
return ret.A
}
func (db *dbRPCServer) StmtNumInput(stID string, ret *Z_DbIntReturn) error {
ret.A = db.dbImpl.StmtNumInput(stID)
return nil
}
type Z_DbStmtQueryArgs struct {
A string
B []driver.NamedValue
}
func (db *dbRPCClient) StmtQuery(stID string, argVals []driver.NamedValue) (string, error) {
args := &Z_DbStmtQueryArgs{
A: stID,
B: argVals,
}
ret := &Z_DbStrErrReturn{}
err := db.client.Call("Plugin.StmtQuery", args, ret)
if err != nil {
log.Printf("error during Plugin.StmtQuery: %v", err)
}
ret.B = decodableError(ret.B)
return ret.A, ret.B
}
func (db *dbRPCServer) StmtQuery(args *Z_DbStmtQueryArgs, ret *Z_DbStrErrReturn) error {
ret.A, ret.B = db.dbImpl.StmtQuery(args.A, args.B)
ret.B = encodableError(ret.B)
return nil
}
func (db *dbRPCClient) StmtExec(stID string, argVals []driver.NamedValue) (ResultContainer, error) {
args := &Z_DbStmtQueryArgs{
A: stID,
B: argVals,
}
ret := &Z_DbResultContErrReturn{}
err := db.client.Call("Plugin.StmtExec", args, ret)
if err != nil {
log.Printf("error during Plugin.StmtExec: %v", err)
}
ret.A.LastIDError = decodableError(ret.A.LastIDError)
ret.A.RowsAffectedError = decodableError(ret.A.RowsAffectedError)
ret.B = decodableError(ret.B)
return ret.A, ret.B
}
func (db *dbRPCServer) StmtExec(args *Z_DbStmtQueryArgs, ret *Z_DbResultContErrReturn) error {
ret.A, ret.B = db.dbImpl.StmtExec(args.A, args.B)
ret.A.LastIDError = encodableError(ret.A.LastIDError)
ret.A.RowsAffectedError = encodableError(ret.A.RowsAffectedError)
ret.B = encodableError(ret.B)
return nil
}
type Z_DbConnArgs struct {
A string
B string
C []driver.NamedValue
}
func (db *dbRPCClient) ConnQuery(connID, q string, argVals []driver.NamedValue) (string, error) {
args := &Z_DbConnArgs{
A: connID,
B: q,
C: argVals,
}
ret := &Z_DbStrErrReturn{}
err := db.client.Call("Plugin.ConnQuery", args, ret)
if err != nil {
log.Printf("error during Plugin.ConnQuery: %v", err)
}
ret.B = decodableError(ret.B)
return ret.A, ret.B
}
func (db *dbRPCServer) ConnQuery(args *Z_DbConnArgs, ret *Z_DbStrErrReturn) error {
ret.A, ret.B = db.dbImpl.ConnQuery(args.A, args.B, args.C)
ret.B = encodableError(ret.B)
return nil
}
type Z_DbResultContErrReturn struct {
A ResultContainer
B error
}
func (db *dbRPCClient) ConnExec(connID, q string, argVals []driver.NamedValue) (ResultContainer, error) {
args := &Z_DbConnArgs{
A: connID,
B: q,
C: argVals,
}
ret := &Z_DbResultContErrReturn{}
err := db.client.Call("Plugin.ConnExec", args, ret)
if err != nil {
log.Printf("error during Plugin.ConnExec: %v", err)
}
ret.A.LastIDError = decodableError(ret.A.LastIDError)
ret.A.RowsAffectedError = decodableError(ret.A.RowsAffectedError)
ret.B = decodableError(ret.B)
return ret.A, ret.B
}
func (db *dbRPCServer) ConnExec(args *Z_DbConnArgs, ret *Z_DbResultContErrReturn) error {
ret.A, ret.B = db.dbImpl.ConnExec(args.A, args.B, args.C)
ret.A.LastIDError = encodableError(ret.A.LastIDError)
ret.A.RowsAffectedError = encodableError(ret.A.RowsAffectedError)
ret.B = encodableError(ret.B)
return nil
}
type Z_DbStrSliceReturn struct {
A []string
}
func (db *dbRPCClient) RowsColumns(rowsID string) []string {
ret := &Z_DbStrSliceReturn{}
err := db.client.Call("Plugin.RowsColumns", rowsID, ret)
if err != nil {
log.Printf("error during Plugin.RowsColumns: %v", err)
}
return ret.A
}
func (db *dbRPCServer) RowsColumns(rowsID string, ret *Z_DbStrSliceReturn) error {
ret.A = db.dbImpl.RowsColumns(rowsID)
return nil
}
func (db *dbRPCClient) RowsClose(resID string) error {
ret := &Z_DbErrReturn{}
err := db.client.Call("Plugin.RowsClose", resID, ret)
if err != nil {
log.Printf("error during Plugin.RowsClose: %v", err)
}
ret.A = decodableError(ret.A)
return ret.A
}
func (db *dbRPCServer) RowsClose(resID string, ret *Z_DbErrReturn) error {
ret.A = db.dbImpl.RowsClose(resID)
ret.A = encodableError(ret.A)
return nil
}
type Z_DbRowScanReturn struct {
A error
B []driver.Value
}
type Z_DbRowScanArg struct {
A string
B []driver.Value
}
func (db *dbRPCClient) RowsNext(rowsID string, dest []driver.Value) error {
args := &Z_DbRowScanArg{
A: rowsID,
B: dest,
}
ret := &Z_DbRowScanReturn{}
err := db.client.Call("Plugin.RowsNext", args, ret)
if err != nil {
log.Printf("error during Plugin.RowsNext: %v", err)
}
ret.A = decodableError(ret.A)
for i, v := range ret.B {
dest[i] = v
}
return ret.A
}
func (db *dbRPCServer) RowsNext(args *Z_DbRowScanArg, ret *Z_DbRowScanReturn) error {
ret.A = db.dbImpl.RowsNext(args.A, args.B)
ret.A = encodableError(ret.A)
// Trick to populate the dest slice. RPC doesn't have a semantic to populate
// pointer type args. So the only way to pass values is via args, and only way
// to return values is via the return struct.
ret.B = args.B
return nil
}
func (db *dbRPCClient) RowsHasNextResultSet(rowsID string) bool {
ret := &Z_DbBoolReturn{}
err := db.client.Call("Plugin.RowsHasNextResultSet", rowsID, ret)
if err != nil {
log.Printf("error during Plugin.RowsHasNextResultSet: %v", err)
}
return ret.A
}
func (db *dbRPCServer) RowsHasNextResultSet(rowsID string, ret *Z_DbBoolReturn) error {
ret.A = db.dbImpl.RowsHasNextResultSet(rowsID)
return nil
}
func (db *dbRPCClient) RowsNextResultSet(rowsID string) error {
ret := &Z_DbErrReturn{}
err := db.client.Call("Plugin.RowsNextResultSet", rowsID, ret)
if err != nil {
log.Printf("error during Plugin.RowsNextResultSet: %v", err)
}
ret.A = decodableError(ret.A)
return ret.A
}
func (db *dbRPCServer) RowsNextResultSet(rowsID string, ret *Z_DbErrReturn) error {
ret.A = db.dbImpl.RowsNextResultSet(rowsID)
ret.A = encodableError(ret.A)
return nil
}
type Z_DbRowsColumnArg struct {
A string
B int
}
func (db *dbRPCClient) RowsColumnTypeDatabaseTypeName(rowsID string, index int) string {
args := &Z_DbRowsColumnArg{
A: rowsID,
B: index,
}
var ret string
err := db.client.Call("Plugin.RowsColumnTypeDatabaseTypeName", args, &ret)
if err != nil {
log.Printf("error during Plugin.RowsColumnTypeDatabaseTypeName: %v", err)
}
return ret
}
func (db *dbRPCServer) RowsColumnTypeDatabaseTypeName(args *Z_DbRowsColumnArg, ret *string) error {
*ret = db.dbImpl.RowsColumnTypeDatabaseTypeName(args.A, args.B)
return nil
}
type Z_DbRowsColumnTypePrecisionScaleReturn struct {
A int64
B int64
C bool
}
func (db *dbRPCClient) RowsColumnTypePrecisionScale(rowsID string, index int) (int64, int64, bool) {
args := &Z_DbRowsColumnArg{
A: rowsID,
B: index,
}
ret := &Z_DbRowsColumnTypePrecisionScaleReturn{}
err := db.client.Call("Plugin.RowsColumnTypePrecisionScale", args, ret)
if err != nil {
log.Printf("error during Plugin.RowsColumnTypePrecisionScale: %v", err)
}
return ret.A, ret.B, ret.C
}
func (db *dbRPCServer) RowsColumnTypePrecisionScale(args *Z_DbRowsColumnArg, ret *Z_DbRowsColumnTypePrecisionScaleReturn) error {
ret.A, ret.B, ret.C = db.dbImpl.RowsColumnTypePrecisionScale(args.A, args.B)
return nil
}

10
plugin/environment.go
Просмотреть файл

@@ -54,17 +54,23 @@ type Environment struct {
logger *mlog.Logger
metrics einterfaces.MetricsInterface
newAPIImpl apiImplCreatorFunc
dbDriver Driver
pluginDir string
webappPluginDir string
prepackagedPlugins []*PrepackagedPlugin
prepackagedPluginsLock sync.RWMutex
}
func NewEnvironment(newAPIImpl apiImplCreatorFunc, pluginDir string, webappPluginDir string, logger *mlog.Logger, metrics einterfaces.MetricsInterface) (*Environment, error) {
func NewEnvironment(newAPIImpl apiImplCreatorFunc,
dbDriver Driver,
pluginDir string, webappPluginDir string,
logger *mlog.Logger,
metrics einterfaces.MetricsInterface) (*Environment, error) {
return &Environment{
logger: logger,
metrics: metrics,
newAPIImpl: newAPIImpl,
dbDriver: dbDriver,
pluginDir: pluginDir,
webappPluginDir: webappPluginDir,
}, nil
@@ -263,7 +269,7 @@ func (env *Environment) Activate(id string) (manifest *model.Manifest, activated
}
if pluginInfo.Manifest.HasServer() {
sup, err := newSupervisor(pluginInfo, env.newAPIImpl(pluginInfo.Manifest), env.logger, env.metrics)
sup, err := newSupervisor(pluginInfo, env.newAPIImpl(pluginInfo.Manifest), env.dbDriver, env.logger, env.metrics)
if err != nil {
return nil, false, errors.Wrapf(err, "unable to start plugin: %v", id)
}

4
plugin/health_check_test.go
Просмотреть файл

@@ -59,7 +59,7 @@ func testPluginHealthCheckSuccess(t *testing.T) {
EnableFile: false,
})
supervisor, err := newSupervisor(bundle, nil, log, nil)
supervisor, err := newSupervisor(bundle, nil, nil, log, nil)
require.NoError(t, err)
require.NotNil(t, supervisor)
defer supervisor.Shutdown()
@@ -106,7 +106,7 @@ func testPluginHealthCheckPanic(t *testing.T) {
EnableFile: false,
})
supervisor, err := newSupervisor(bundle, nil, log, nil)
supervisor, err := newSupervisor(bundle, nil, nil, log, nil)
require.NoError(t, err)
require.NotNil(t, supervisor)
defer supervisor.Shutdown()

54
plugin/helpers.go
Просмотреть файл

@@ -4,6 +4,8 @@
package plugin
import (
"database/sql/driver"
"github.com/mattermost/mattermost-server/v5/model"
)
@@ -95,3 +97,55 @@ type Helpers interface {
type HelpersImpl struct {
API API
}
// ResultContainer contains the output from the LastInsertID
// and RowsAffected methods for a given set of rows.
// It is used to embed another round-trip to the server,
// and helping to avoid tracking results on the server.
type ResultContainer struct {
LastID int64
LastIDError error
RowsAffected int64
RowsAffectedError error
}
type Driver interface {
// Connection
Conn() (string, error)
ConnPing(connID string) error
ConnClose(connID string) error
ConnQuery(connID, q string, args []driver.NamedValue) (string, error) // rows
ConnExec(connID, q string, args []driver.NamedValue) (ResultContainer, error) // result
// Transaction
Tx(connID string, opts driver.TxOptions) (string, error)
TxCommit(txID string) error
TxRollback(txID string) error
// Statement
Stmt(connID, q string) (string, error)
StmtClose(stID string) error
StmtNumInput(stID string) int
StmtQuery(stID string, args []driver.NamedValue) (string, error) // rows
StmtExec(stID string, args []driver.NamedValue) (ResultContainer, error) // result
// Rows
RowsColumns(rowsID string) []string
RowsClose(rowsID string) error
RowsNext(rowsID string, dest []driver.Value) error
RowsHasNextResultSet(rowsID string) bool
RowsNextResultSet(rowsID string) error
RowsColumnTypeDatabaseTypeName(rowsID string, index int) string
RowsColumnTypePrecisionScale(rowsID string, index int) (int64, int64, bool)
// TODO: add this
// RowsColumnScanType(rowsID string, index int) reflect.Type
// Note: the following cannot be implemented because either MySQL or PG
// does not support it. So this implementation has to be a common subset
// of both DB implementations.
// RowsColumnTypeLength(rowsID string, index int) (int64, bool)
// RowsColumnTypeNullable(rowsID string, index int) (bool, bool)
// ResetSession(ctx context.Context) error
// IsValid() bool
}

7
plugin/supervisor.go
Просмотреть файл

@@ -27,7 +27,7 @@ type supervisor struct {
pid int
}
func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, parentLogger *mlog.Logger, metrics einterfaces.MetricsInterface) (retSupervisor *supervisor, retErr error) {
func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, driver Driver, parentLogger *mlog.Logger, metrics einterfaces.MetricsInterface) (retSupervisor *supervisor, retErr error) {
sup := supervisor{}
defer func() {
if retErr != nil {
@@ -44,8 +44,9 @@ func newSupervisor(pluginInfo *model.BundleInfo, apiImpl API, parentLogger *mlog
pluginMap := map[string]plugin.Plugin{
"hooks": &hooksPlugin{
log: wrappedLogger,
apiImpl: &apiTimerLayer{pluginInfo.Manifest.Id, apiImpl, metrics},
log: wrappedLogger,
driverImpl: driver,
apiImpl: &apiTimerLayer{pluginInfo.Manifest.Id, apiImpl, metrics},
},
}

6
plugin/supervisor_test.go
Просмотреть файл

@@ -41,7 +41,7 @@ func testSupervisorInvalidExecutablePath(t *testing.T) {
ConsoleLevel: "error",
EnableFile: false,
})
supervisor, err := newSupervisor(bundle, nil, log, nil)
supervisor, err := newSupervisor(bundle, nil, nil, log, nil)
assert.Nil(t, supervisor)
assert.Error(t, err)
}
@@ -60,7 +60,7 @@ func testSupervisorNonExistentExecutablePath(t *testing.T) {
ConsoleLevel: "error",
EnableFile: false,
})
supervisor, err := newSupervisor(bundle, nil, log, nil)
supervisor, err := newSupervisor(bundle, nil, nil, log, nil)
require.Error(t, err)
require.Nil(t, supervisor)
}
@@ -90,7 +90,7 @@ func testSupervisorStartTimeout(t *testing.T) {
ConsoleLevel: "error",
EnableFile: false,
})
supervisor, err := newSupervisor(bundle, nil, log, nil)
supervisor, err := newSupervisor(bundle, nil, nil, log, nil)
require.Error(t, err)
require.Nil(t, supervisor)
}