adapter

README

中继适配器SDK

按照 QCP 协议规范封装适配SDK，以实现安全、简便和快捷的接入非 Tendermint 技术栈的区块链。

为接入链提供交易发布功能，并为中继通信提供标准接口及相关服务（ Http Rpc、Web Socket ）。

Documentation

Overview

Package adapter 按照 QCP 协议规范封装适配SDK，以实现安全、简便和快捷的接入非 Tendermint 技术栈的区块链。

为接入链提供交易发布功能，并为中继通信提供标准接口及相关服务（ Http Rpc、Web Socket ）。

Index

• Variables
• type BaseService
  • func NewBaseService(name string, impl Service) *BaseService
  • func (bs *BaseService) IsRunning() bool
  • func (bs *BaseService) OnReset() error
  • func (bs *BaseService) OnStart() error
  • func (bs *BaseService) OnStop()
  • func (bs *BaseService) Quit() <-chan struct{}
  • func (bs *BaseService) Reset() error
  • func (bs *BaseService) Start() error
  • func (bs *BaseService) Stop() error
  • func (bs *BaseService) String() string
  • func (bs *BaseService) Wait()
• type Broadcaster
• type DefaultBroadcaster
  • func (b *DefaultBroadcaster) BroadcastTx(tx *txs.TxQcp) (err error)
• type HandlerService
• type Option
  • func BufferCapacity(cap int) Option
• type Query
• type Receiver
• type RemoteAdapter
  • func NewRemoteAdapter(name, id, listenAddr string, r Receiver, b Broadcaster) (*RemoteAdapter, error)
• type Server
  • func NewServer(options ...Option) *Server
  • func (s *Server) BufferCapacity() int
  • func (s *Server) OnReset() error
  • func (s *Server) OnStart() error
  • func (s *Server) OnStop()
  • func (s *Server) Publish(ctx context.Context, msg interface{}) error
  • func (s *Server) PublishWithTags(ctx context.Context, msg interface{}, tags TagMap) error
  • func (s *Server) Subscribe(ctx context.Context, clientID string, query Query, out chan<- interface{}) error
  • func (s *Server) Unsubscribe(ctx context.Context, clientID string, query Query) error
  • func (s *Server) UnsubscribeAll(ctx context.Context, clientID string) error
• type Service
• type TagMap
  • func NewTagMap(data map[string]string) TagMap

Variables

var (
	// ErrSubscriptionNotFound is returned when a client tries to unsubscribe
	// from not existing subscription.
	ErrSubscriptionNotFound = errors.New("subscription not found")

	// ErrAlreadySubscribed is returned when a client tries to subscribe twice or
	// more using the same query.
	ErrAlreadySubscribed = errors.New("already subscribed")
)

var (
	// ErrAlreadyStarted 已启动错误
	ErrAlreadyStarted = errors.New("already started")
	// ErrAlreadyStopped 已停止错误
	ErrAlreadyStopped = errors.New("already stopped")
)

Types

type BaseService

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

BaseService 封装服务基础操作方法

func NewBaseService

func NewBaseService(name string, impl Service) *BaseService

NewBaseService creates a new BaseService.

func (*BaseService) IsRunning

func (bs *BaseService) IsRunning() bool

IsRunning implements Service by returning true or false depending on the service's state.

func (*BaseService) OnReset

func (bs *BaseService) OnReset() error

OnReset implements Service by panicking.

func (*BaseService) OnStart

func (bs *BaseService) OnStart() error

OnStart implements Service by doing nothing. NOTE: Do not put anything in here, that way users don't need to call BaseService.OnStart()

func (*BaseService) OnStop

func (bs *BaseService) OnStop()

OnStop implements Service by doing nothing. NOTE: Do not put anything in here, that way users don't need to call BaseService.OnStop()

func (*BaseService) Quit

func (bs *BaseService) Quit() <-chan struct{}

Quit Implements Service by returning a quit channel.

func (*BaseService) Reset

func (bs *BaseService) Reset() error

Reset implements Service by calling OnReset callback (if defined). An error will be returned if the service is running.

func (*BaseService) Start

func (bs *BaseService) Start() error

Start implements Service by calling OnStart (if defined). An error will be returned if the service is already running or stopped. Not to start the stopped service, you need to call Reset.

func (*BaseService) Stop

func (bs *BaseService) Stop() error

Stop implements Service by calling OnStop (if defined) and closing quit channel. An error will be returned if the service is already stopped.

func (*BaseService) String

func (bs *BaseService) String() string

String implements Servce by returning a string representation of the service.

func (*BaseService) Wait

func (bs *BaseService) Wait()

Wait blocks until the service is stopped.

type Broadcaster

type Broadcaster interface {
	BroadcastTx(tx *txs.TxQcp) error
}

Broadcaster 交易广播接口，通过该接口广播的交易即表示需要通过中继跨链提交交易以最终完成交易。

type DefaultBroadcaster

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

DefaultBroadcaster 实现内存交易广播器。

作为接入链跨链交易的缓存以提高查询的相关功能的执行效率。 交易在广播器中不会缓存，而是直接转发给中继适配服务。

func (*DefaultBroadcaster) BroadcastTx

func (b *DefaultBroadcaster) BroadcastTx(tx *txs.TxQcp) (err error)

BroadcastTx 实现交易广播接口，调用响应的交易及交易事件发布接口，以通过中继跨链提交交易以最终完成交易。

因为按照 QCP 协议规范定义，中继都是在接收到交易事件后查询交易数据，因此应保证先调用发布交易接口，然后再调用发布事件接口。

type HandlerService

type HandlerService interface {
	Start() error
	Stop() error
	GetCodec() *amino.Codec
	PublishTx(tx *txs.TxQcp) error
	PublishEvent(e *tmtypes.EventDataTx) error
}

HandlerService 中继基础服务封装接口

type Option

type Option func(*Server)

Option sets a parameter for the server.

func BufferCapacity

func BufferCapacity(cap int) Option

BufferCapacity allows you to specify capacity for the internal server's queue. Since the server, given Y subscribers, could only process X messages, this option could be used to survive spikes (e.g. high amount of transactions during peak hours).

type Query

type Query interface {
	Matches(tags TagMap) bool
	String() string
}

Query defines an interface for a query to be used for subscribing.

type Receiver

type Receiver interface {
	ReceiveTx(tx *txs.TxQcp) error
}

Receiver 交易接收接口，接收中继从其他接入链发来的跨链交易。

type RemoteAdapter

type RemoteAdapter struct {
	HandlerService
	Broadcaster
	Receiver
}

RemoteAdapter 适配接口封装，封装交易广播接口和交易接收接口。

交易广播接口（ Broadcaster ）为调用端检测到交易事件时，由调用端调用; 交易接收接口（ Receiver ）为中继适配服务接收到远端中继广播的交易后，由适配服务回调通知调用方接收到远端跨链交易。

func NewRemoteAdapter

func NewRemoteAdapter(name, id, listenAddr string, r Receiver, b Broadcaster) (*RemoteAdapter, error)

NewRemoteAdapter 创建新的交易广播器

type Server

type Server struct {
	BaseService
	// contains filtered or unexported fields
}

Server allows clients to subscribe/unsubscribe for messages, publishing messages with or without tags, and manages internal state.

func NewServer

func NewServer(options ...Option) *Server

NewServer returns a new server. See the commentary on the Option functions for a detailed description of how to configure buffering. If no options are provided, the resulting server's queue is unbuffered.

func (*Server) BufferCapacity

func (s *Server) BufferCapacity() int

BufferCapacity returns capacity of the internal server's queue.

func (*Server) OnReset

func (s *Server) OnReset() error

OnReset implements Service.OnReset

func (*Server) OnStart

func (s *Server) OnStart() error

OnStart implements Service.OnStart by starting the server.

func (*Server) OnStop

func (s *Server) OnStop()

OnStop implements Service.OnStop by shutting down the server.

func (*Server) Publish

func (s *Server) Publish(ctx context.Context, msg interface{}) error

Publish publishes the given message. An error will be returned to the caller if the context is canceled.

func (*Server) PublishWithTags

func (s *Server) PublishWithTags(ctx context.Context, msg interface{}, tags TagMap) error

PublishWithTags publishes the given message with the set of tags. The set is matched with clients queries. If there is a match, the message is sent to the client.

func (*Server) Subscribe

func (s *Server) Subscribe(ctx context.Context, clientID string, query Query, out chan<- interface{}) error

Subscribe creates a subscription for the given client. It accepts a channel on which messages matching the given query can be received. An error will be returned to the caller if the context is canceled or if subscription already exist for pair clientID and query.

func (*Server) Unsubscribe

func (s *Server) Unsubscribe(ctx context.Context, clientID string, query Query) error

Unsubscribe removes the subscription on the given query. An error will be returned to the caller if the context is canceled or if subscription does not exist.

func (*Server) UnsubscribeAll

func (s *Server) UnsubscribeAll(ctx context.Context, clientID string) error

UnsubscribeAll removes all client subscriptions. An error will be returned to the caller if the context is canceled or if subscription does not exist.

type Service

type Service interface {
	// Start the service.
	// If it's already started or stopped, will return an error.
	// If OnStart() returns an error, it's returned by Start()
	Start() error
	OnStart() error

	// Stop the service.
	// If it's already stopped, will return an error.
	// OnStop must never error.
	Stop() error
	OnStop()

	// Reset the service.
	// Panics by default - must be overwritten to enable reset.
	Reset() error
	OnReset() error

	// Return true if the service is running
	IsRunning() bool

	// Quit returns a channel, which is closed once service is stopped.
	Quit() <-chan struct{}

	// String representation of the service
	String() string
}

Service defines a service that can be started, stopped, and reset.

type TagMap

type TagMap interface {
	// Get returns the value for a key, or nil if no value is present.
	// The ok result indicates whether value was found in the tags.
	Get(key string) (value string, ok bool)
	// Len returns the number of tags.
	Len() int
}

TagMap is used to associate tags to a message. They can be queried by subscribers to choose messages they will received.

func NewTagMap

func NewTagMap(data map[string]string) TagMap

NewTagMap constructs a new immutable tag set from a map.