common

Documentation

Index

• Constants
• type Caller
• type CallerPlugin
  • func (CallerPlugin) Client(b *plugin.MuxBroker, c *rpc.Client) (interface{}, error)
  • func (p *CallerPlugin) Server(*plugin.MuxBroker) (interface{}, error)
• type CallerRPC
  • func (c *CallerRPC) Commit(request RpcReqComponent) RpcResComponent
  • func (c *CallerRPC) Execute(request RpcReqComponent) RpcResComponent
  • func (c *CallerRPC) Prepare(request RpcReqComponent) RpcResComponent
  • func (c *CallerRPC) Rollback(request RpcReqComponent) RpcResComponent
• type CallerRPCServer
  • func (c *CallerRPCServer) Commit(req RpcReqComponent, resp *RpcResComponent) error
  • func (c *CallerRPCServer) Execute(req RpcReqComponent, resp *RpcResComponent) error
  • func (c *CallerRPCServer) Prepare(req RpcReqComponent, resp *RpcResComponent) error
  • func (c *CallerRPCServer) Rollback(req RpcReqComponent, resp *RpcResComponent) error
• type Component
  • func NewComponent(executeHandler interface{}, prepareHandler interface{}, ...) *Component
  • func (component *Component) Commit(config []byte, input []byte, output []byte, context Context) ([]byte, error)
  • func (component *Component) Execute(config []byte, input []byte, context Context) ([]byte, error)
  • func (component *Component) Prepare(config []byte, input []byte, context Context) ([]byte, error)
  • func (component *Component) Rollback(config []byte, input []byte, output []byte, context Context) ([]byte, error)
• type ComponentFileInfo
• type Context
  • func (c *Context) ReadFile(path string) (io.ReadCloser, error)
  • func (c *Context) WriteFile(reader io.Reader) (string, error)
• type DiskInfo
• type FTPInfo
• type Handler
  • func NewCommitAndRollbackHandler(handlerFunc interface{}) Handler
  • func NewExecuteAndPrepareHandler(handlerFunc interface{}) Handler
• type MinIOInfo
• type RpcComponentContext
• type RpcReqComponent
• type RpcResComponent

Constants

const (
	DefaultComponentName = "default"
	MagicCookieKey       = "BASIC_PLUGIN"
	MagicCookieValue     = "ipaas-go-component"
)

Types

type Caller

type Caller interface {
	Execute(request RpcReqComponent) RpcResComponent

	Prepare(request RpcReqComponent) RpcResComponent

	Rollback(request RpcReqComponent) RpcResComponent

	Commit(request RpcReqComponent) RpcResComponent
}

Caller is the interface that we're exposing as a plugin.

type CallerPlugin

type CallerPlugin struct {
	// Impl Injection
	Impl Caller
}

This is the implementation of plugin.Plugin so we can serve/consume this

This has two methods: Server must return an RPC server for this plugin type. We construct a GreeterRPCServer for this.

Client must return an implementation of our interface that communicates over an RPC client. We return GreeterRPC for this.

Ignore MuxBroker. That is used to create more multiplexed streams on our plugin connection and is a more advanced use case.

func (CallerPlugin) Client

func (CallerPlugin) Client(b *plugin.MuxBroker, c *rpc.Client) (interface{}, error)

func (*CallerPlugin) Server

func (p *CallerPlugin) Server(*plugin.MuxBroker) (interface{}, error)

type CallerRPC

type CallerRPC struct {
	// contains filtered or unexported fields
}

Here is an implementation that talks over RPC

func (*CallerRPC) Commit

func (c *CallerRPC) Commit(request RpcReqComponent) RpcResComponent

func (*CallerRPC) Execute

func (c *CallerRPC) Execute(request RpcReqComponent) RpcResComponent

func (*CallerRPC) Prepare

func (c *CallerRPC) Prepare(request RpcReqComponent) RpcResComponent

func (*CallerRPC) Rollback

func (c *CallerRPC) Rollback(request RpcReqComponent) RpcResComponent

type CallerRPCServer

type CallerRPCServer struct {
	// This is the real implementation
	Impl Caller
}

Here is the RPC server that GreeterRPC talks to, conforming to the requirements of net/rpc

func (*CallerRPCServer) Commit

func (c *CallerRPCServer) Commit(req RpcReqComponent, resp *RpcResComponent) error

func (*CallerRPCServer) Execute

func (c *CallerRPCServer) Execute(req RpcReqComponent, resp *RpcResComponent) error

func (*CallerRPCServer) Prepare

func (c *CallerRPCServer) Prepare(req RpcReqComponent, resp *RpcResComponent) error

func (*CallerRPCServer) Rollback

func (c *CallerRPCServer) Rollback(req RpcReqComponent, resp *RpcResComponent) error

type Component

type Component struct {
	FileClient fileClient
	// contains filtered or unexported fields
}

func NewComponent

func NewComponent(executeHandler interface{}, prepareHandler interface{},
	commitHandler interface{}, rollbackHandler interface{}) *Component

NewFunction which creates a Function with a given Handler

func (*Component) Commit

func (component *Component) Commit(config []byte, input []byte, output []byte, context Context) ([]byte, error)

*

• 组件事务提交

func (*Component) Execute

func (component *Component) Execute(config []byte, input []byte, context Context) ([]byte, error)

*

• 调度执行方法

func (*Component) Prepare

func (component *Component) Prepare(config []byte, input []byte, context Context) ([]byte, error)

*

• 调度执行方法

func (*Component) Rollback

func (component *Component) Rollback(config []byte, input []byte, output []byte, context Context) ([]byte, error)

*

• 组件

type ComponentFileInfo

type ComponentFileInfo struct {
	Type  string    `json:"type"`
	Disk  DiskInfo  `json:"disk"`
	FTP   FTPInfo   `json:"ftp"`
	MinIO MinIOInfo `json:"minio"`
}

type Context

type Context struct {
	/**
	 * 子组件名称
	 */
	Name string `json:"name"`
	/**
	 * 子组件版本
	 */
	Version string `json:"version"`
	/**
	 * 设置事务ID, commit和rollback可以获取使用
	 */
	TransactionId string `json:"transactionId"`
	/**
	 * 请求id，用于链路追踪
	 */
	RequestId string `json:"requestId"`

	Logger hclog.Logger

	FileClient fileClient
}

func (*Context) ReadFile

func (c *Context) ReadFile(path string) (io.ReadCloser, error)

读取文件, path文件路径，返回文件流或者错误

func (*Context) WriteFile

func (c *Context) WriteFile(reader io.Reader) (string, error)

写入文件, reader文件流，返回文件路径或者错误

type DiskInfo

type DiskInfo struct {
	RootDir string `json:"rootDir"`
}

type FTPInfo

type FTPInfo struct {
	Host     string `json:"host"`
	Username string `json:"username"`
	Password string `json:"password"`
	RootDir  string `json:"rootDir"`
}

type Handler

type Handler interface {
	/**
	 * 执行逻辑
	 */
	Invoke(config []byte, input []byte, output []byte, context Context) ([]byte, error)
}

func NewCommitAndRollbackHandler

func NewCommitAndRollbackHandler(handlerFunc interface{}) Handler

NewHandler creates a base fc component from the given handler function. The returned Handler performs JSON serialization and deserialization, and delegates to the input handler function. The handler function parameter must satisfy the rules documented by Start. If handlerFunc is not a valid handler, the returned Handler simply reports the validation error.

func NewExecuteAndPrepareHandler

func NewExecuteAndPrepareHandler(handlerFunc interface{}) Handler

NewHandler creates a base fc component from the given handler function. The returned Handler performs JSON serialization and deserialization, and delegates to the input handler function. The handler function parameter must satisfy the rules documented by Start. If handlerFunc is not a valid handler, the returned Handler simply reports the validation error.

type MinIOInfo

type MinIOInfo struct {
	Endpoint  string `json:"endpoint"`
	AccessKey string `json:"accessKey"`
	SecretKey string `json:"secretKey"`
	Bucket    string `json:"bucket"`
}

type RpcComponentContext

type RpcComponentContext struct {
	/**
	 * 子组件名称
	 */
	Name string `json:"name"`
	/**
	 * 子组件版本
	 */
	Version string `json:"version"`
	/**
	 * 设置事务ID, commit和rollback可以获取使用
	 */
	TransactionId string `json:"transactionId"`
	/**
	 * 请求id，用于链路追踪
	 */
	RequestId string `json:"requestId"`
}

type RpcReqComponent

type RpcReqComponent struct {
	Config           []byte              `json:"config"`
	Input            []byte              `json:"input"`
	Output           []byte              `json:"output"`
	ComponentContext RpcComponentContext `json:"componentContext"`
}

type RpcResComponent

type RpcResComponent struct {
	Output       []byte `json:"output"`
	ErrorMessage string `json:"errorMessage"`
}