pubsubtaskplugin

README

pubsub-task-plugin-framework

Cloud pub/sub を利用した並列分散型のタスク実行フレームワークです。 タスクは Plugin で管理することが可能で、publisher から subscriber で実行するプラグインを指定することが出来ます。

Structure

+ publisher.go
+ subscriber.go
+ lib/
   `- protocol.go
+ samples/
   |- sample-publisher-client.go
   |- sample-subscriber-client.go
   `- plugins
        `- hello.go

How to use

Prepare

setup environment

$ git clone git@github.com:m-yoshimo/pubsub-task-plugin-framework.git
$ export GOOGLE_PROJECT_ID="xxxx"

get libraries

$ go get cloud.google.com/go/pubsub
$ go get google.golang.org/api/iterator
$ go get github.com/dullgiulio/pingo
$ go get github.com/m-yoshimo/pubsub-task-plugin-framework

Implement of your plugin

plugin は CreateMessage と Exec の 2 つの関数を実装します。 また、plugin を利用可能にするために、main 関数で plugin 登録を行います。 サンプルとして、publisher で名前を指定して、subscriber で hello + ${name} を表示する HelloPlugin (./plugins/hello.go) を用視しています。 実装の参考にどうぞ。

CreateMesaage(args map[string]string, msg *string) error

pub/sub メッセージを生成します。

• args: subscriber で実行する際の引数です (Exec 参照)
• msg: メッセージを格納するポインタです

たいていの場合、サンプルの HelloPlugin (./plugin/hello.go) と同じ実装で問題ないはずです。

Exec(args map[string]string, res *string) error

subscriber で実行するタスクを実装します。

• args: publisher から指定された引数です
• res: Exec のレスポンスを指定しますが、現状は利用していないので無視して構いません

Documentation

Index

• func ComposePluginMessage(name string, binName string, args map[string]string) (string, error)
• func CreateStopMessage() (string, error)
• func ParseStopMessage(data []byte) error
• type Message
• type Plugin
  • func ParsePluginMessage(data []byte, pluginDir string) (*Plugin, error)
• type Publisher
  • func (p *Publisher) Do(proj string, topicName string, plugin *Plugin) error
  • func (p *Publisher) NewPlugin(name string, path string, args map[string]string) (*Plugin, error)
• type Subscriber
  • func (s *Subscriber) Do(proj string, topicName string, subName string, concurrency int) error
  • func (s *Subscriber) GetChannelCapacity() int
  • func (s *Subscriber) GetChannelLength() int

Functions

func ComposePluginMessage

func ComposePluginMessage(name string, binName string, args map[string]string) (string, error)

ComposePluginMessage creates message of pub/sub

func CreateStopMessage

func CreateStopMessage() (string, error)

CreateStopMessage creates message to stop subscriber

func ParseStopMessage

func ParseStopMessage(data []byte) error

ParseStopMessage parses message of stop from publisher

Types

type Message

type Message struct {
	Type    string            `json:"type"`
	Command string            `json:"command"`
	Name    string            `json:"name"`
	BinName string            `json:"binName"`
	Args    map[string]string `json:"args"`
}

Message format of pub/sub TODO: 構造体を外部に隠蔽する方法を調べる

type Plugin

type Plugin struct {
	Name string
	Path string
	Args map[string]string
}

Plugin struct of information to execute remote plugin

func ParsePluginMessage

func ParsePluginMessage(data []byte, pluginDir string) (*Plugin, error)

ParsePluginMessage parses message of pub/sub

type Publisher

type Publisher struct{}

Publisher for plugin

func (*Publisher) Do

func (p *Publisher) Do(proj string, topicName string, plugin *Plugin) error

Do Publisher plugin

func (*Publisher) NewPlugin

func (p *Publisher) NewPlugin(name string, path string, args map[string]string) (*Plugin, error)

NewPlugin creates Publisher plugin

type Subscriber

type Subscriber struct {
	PluginDir string
}

Subscriber for plugin

func (*Subscriber) Do

func (s *Subscriber) Do(proj string, topicName string, subName string, concurrency int) error

Do Subscriber plugin

func (*Subscriber) GetChannelCapacity

func (s *Subscriber) GetChannelCapacity() int

GetChannelCapacity capacity of channel in subscriber

func (*Subscriber) GetChannelLength

func (s *Subscriber) GetChannelLength() int

GetChannelLength length of channel in subscriber