feat(*) new methods

- SetPluginDir()
- GetPluginInfo()
- StartInstance()
- CloseInstance()
- Assume multiple threads; add mutex
This commit is contained in:
Javier Guerra
2019-11-15 12:40:50 -05:00
committed by Guilherme Salazar
parent 25ad77318d
commit 09ddf7101c

View File

@@ -1,6 +1,7 @@
package main package main
import ( import (
"encoding/json"
"flag" "flag"
"fmt" "fmt"
"github.com/kong/go-pdk" "github.com/kong/go-pdk"
@@ -12,6 +13,7 @@ import (
"plugin" "plugin"
"reflect" "reflect"
"strings" "strings"
"sync"
) )
var socket = flag.String("socket", "", "Socket to listen into") var socket = flag.String("socket", "", "Socket to listen into")
@@ -19,18 +21,15 @@ var socket = flag.String("socket", "", "Socket to listen into")
func runServer(listener net.Listener) { func runServer(listener net.Listener) {
var handle codec.MsgpackHandle var handle codec.MsgpackHandle
go func() { for {
for { conn, err := listener.Accept()
conn, err := listener.Accept() if err != nil {
if err != nil { log.Printf("accept(): %s", err)
log.Printf("accept(): %s", err) return
return
}
// rpcCodec := codec.GoRpc.ServerCodec(conn, &handle)
rpcCodec := codec.MsgpackSpecRpc.ServerCodec(conn, &handle)
rpc.ServeCodec(rpcCodec)
} }
}() rpcCodec := codec.MsgpackSpecRpc.ServerCodec(conn, &handle)
rpc.ServeCodec(rpcCodec)
}
} }
func main() { func main() {
@@ -43,7 +42,7 @@ func main() {
return return
} }
rpc.RegisterName("plugin", &PluginServer{}) rpc.RegisterName("plugin", newServer())
runServer(listener) runServer(listener)
} }
@@ -51,8 +50,9 @@ func main() {
// --- pluginData --- // // --- pluginData --- //
type pluginData struct { type pluginData struct {
name string
code *plugin.Plugin code *plugin.Plugin
initialized bool constructor func() interface{}
config interface{} config interface{}
handlers map[string]func(kong *pdk.PDK) handlers map[string]func(kong *pdk.PDK)
} }
@@ -69,7 +69,7 @@ type (
func (plug *pluginData) setHandlers() { func (plug *pluginData) setHandlers() {
handlers := map[string]func(kong *pdk.PDK){} handlers := map[string]func(kong *pdk.PDK){}
config := plug.config config := plug.constructor()
if h, ok := config.(certificater); ok { handlers["certificate"] = h.Certificate } if h, ok := config.(certificater); ok { handlers["certificate"] = h.Certificate }
if h, ok := config.(rewriter) ; ok { handlers["rewrite"] = h.Rewrite } if h, ok := config.(rewriter) ; ok { handlers["rewrite"] = h.Rewrite }
@@ -82,21 +82,45 @@ func (plug *pluginData) setHandlers() {
plug.handlers = handlers plug.handlers = handlers
} }
// --- instanceData --- //
type instanceData struct {
id int
plugin *pluginData
initialized bool
config interface{}
ipc chan string
pdk *pdk.PDK
}
// --- PluginServer --- // // --- PluginServer --- //
type PluginServer struct { type PluginServer struct {
lock sync.RWMutex
pluginsDir string pluginsDir string
plugins map[string]pluginData nextId int
plugins map[string]*pluginData
instances map[int]instanceData
}
func newServer() *PluginServer {
return &PluginServer{
plugins: map[string]*pluginData{},
instances: map[int]instanceData{},
}
} }
/// exported method /// exported method
func (s *PluginServer) SetPluginDir(dir string, reply *string) error { func (s *PluginServer) SetPluginDir(dir string, reply *string) error {
s.lock.Lock()
s.pluginsDir = dir s.pluginsDir = dir
s.lock.Unlock()
*reply = "ok" *reply = "ok"
return nil return nil
} }
func (s PluginServer) loadPlugin(name string) (plug pluginData, err error) { func (s PluginServer) loadPlugin(name string) (plug *pluginData, err error) {
s.lock.RLock()
plug, ok := s.plugins[name] plug, ok := s.plugins[name]
s.lock.RUnlock()
if ok { if ok {
return return
} }
@@ -112,19 +136,25 @@ func (s PluginServer) loadPlugin(name string) (plug pluginData, err error) {
err = fmt.Errorf("No constructor function on plugin %s: %w", name, err) err = fmt.Errorf("No constructor function on plugin %s: %w", name, err)
return return
} }
constructor, ok := constructorSymbol.(func() interface{}) constructor, ok := constructorSymbol.(func() interface{})
if !ok { if !ok {
err = fmt.Errorf("Wrong constructor signature on plugin %s: %w", name, err) err = fmt.Errorf("Wrong constructor signature on plugin %s: %w", name, err)
return return
} }
plug = pluginData{ plug = &pluginData{
code: code, name: name,
config: constructor(), code: code,
constructor: constructor,
config: constructor(),
} }
plug.setHandlers() plug.setHandlers()
s.lock.Lock()
s.plugins[name] = plug s.plugins[name] = plug
s.lock.Unlock()
return return
} }
@@ -185,39 +215,37 @@ func getSchemaType(t reflect.Type) (string, bool) {
} }
type PluginInfo struct { type PluginInfo struct {
name string Name string
phases []string Phases []string
version string Version string
priority int Priority int
schema string Schema string
} }
/// exported method /// exported method
func (s *PluginServer) GetPluginInfo(name string, info *PluginInfo) error { func (s PluginServer) GetPluginInfo(name string, info *PluginInfo) error {
info.name = name
plug, err := s.loadPlugin(name) plug, err := s.loadPlugin(name)
if err != nil { if err != nil {
return err return err
} }
*info = PluginInfo{} *info = PluginInfo{Name: name}
info.phases = make([]string, len(plug.handlers)) info.Phases = make([]string, len(plug.handlers))
var i = 0 var i = 0
for name := range plug.handlers { for name := range plug.handlers {
info.phases[i] = name info.Phases[i] = name
i++ i++
} }
v, _ := plug.code.Lookup("Version") v, _ := plug.code.Lookup("Version")
if v != nil { if v != nil {
info.version = v.(string) info.Version = v.(string)
} }
prio, _ := plug.code.Lookup("Priority") prio, _ := plug.code.Lookup("Priority")
if prio != nil { if prio != nil {
info.priority = prio.(int) info.Priority = prio.(int)
} }
var out strings.Builder var out strings.Builder
@@ -230,7 +258,75 @@ func (s *PluginServer) GetPluginInfo(name string, info *PluginInfo) error {
out.WriteString(`}}]}`) out.WriteString(`}}]}`)
info.schema = out.String() info.Schema = out.String()
return nil
}
type PluginConfig struct {
Name string
Config []byte
}
type InstanceStatus struct {
Name string
Id int
Config interface{}
}
/// exported method
func (s *PluginServer) StartInstance(config PluginConfig, status *InstanceStatus) error {
plug, err := s.loadPlugin(config.Name)
if err != nil {
return err
}
instanceConfig := plug.constructor()
if err := json.Unmarshal(config.Config, instanceConfig); err != nil {
return fmt.Errorf("Decoding config: %w", err)
}
instance := instanceData{
id: s.nextId,
plugin: plug,
config: instanceConfig,
}
s.lock.Lock()
s.nextId++
s.instances[instance.id] = instance
s.lock.Unlock()
*status = InstanceStatus{
Name: config.Name,
Id: instance.id,
Config: instance.config,
}
return nil
}
/// exported method
func (s PluginServer) CloseInstance(id int, status *InstanceStatus) error {
s.lock.RLock()
instance, ok := s.instances[id]
s.lock.RUnlock()
if !ok {
return fmt.Errorf("No plugin instance %d", id)
}
*status = InstanceStatus{
Name: instance.plugin.name,
Id: instance.id,
Config: instance.config,
}
// kill?
s.lock.Lock()
delete(s.instances, id)
s.lock.Unlock()
return nil return nil
} }