变更记录

11/23/2020：初始草案
10/06/2022：引入基于 hashicorp/go-plugin 的插件系统
10/14/2022：
- 添加 ListenCommit，将一个区块中的状态写入压平成单个批次。
- 从缓存存储中移除监听器，只应监听 rootmulti.Store。
- 移除 HaltAppOnDeliveryError()，错误默认会向上传播；如果实现不希望传播错误，应返回 nil。
26/05/2023：更新为 ABCI 2.0

状态

提议中

摘要

本 ADR 定义了一组变更，用于支持监听各个 KVStore 的状态变化，并将这些数据暴露给消费者。

背景

当前，KVStore 数据可以通过查询进行远程访问，这些查询要么通过 Tendermint 和 ABCI 处理，要么通过 gRPC 服务器处理。除了这种请求/响应式查询之外，如果还能有一种方式在状态变化发生时实时监听这些变化，将会很有价值。

决策

我们将修改 CommitMultiStore 接口及其具体的 rootmulti 实现，并引入新的 listenkv.Store，以支持监听底层 KVStore 中的状态变化。我们不需要监听缓存存储，因为无法确定这些写入最终一定会被提交，而且这些写入最终也会在 rootmulti.Store 中重复出现，因此我们只应监听 rootmulti.Store。我们将引入一个插件系统，用于配置和运行流式服务，把这些状态变化及其周边的 ABCI 消息上下文写入不同的目标位置。

监听

在新文件 store/types/listening.go 中，我们将创建一个 MemoryListener 结构体，用于从 KVStore 中流式输出采用 protobuf 编码的 KV 键值对状态变化。 MemoryListener 将由具体的 rootmulti 实现在内部使用，用于收集来自 KVStore 的状态变化。

// MemoryListener 监听状态写入，并在内存中累积记录。
type MemoryListener struct {
    stateCache []StoreKVPair
}

// NewMemoryListener 创建一个在内存中累积状态写入的监听器。
func NewMemoryListener() *MemoryListener {
    return &MemoryListener{
}
}

// OnWrite 将状态变化事件写入内部缓存
func (fl *MemoryListener)

OnWrite(storeKey StoreKey, key []byte, value []byte, delete bool) {
    fl.stateCache = append(fl.stateCache, StoreKVPair{
    StoreKey: storeKey.Name(),
    Delete:   delete,
    Key:      key,
    Value:    value,
})
}

// PopStateCache 返回当前状态缓存并将其置为 nil
func (fl *MemoryListener)

PopStateCache() []StoreKVPair {
    res := fl.stateCache
	fl.stateCache = nil
	return res
}

我们还将为这些 KV 键值对定义一个 protobuf 类型。除了 key 和 value 字段之外，这个消息还会包含源 KVStore 的 StoreKey，这样我们就可以汇总来自不同 KVStore 的信息，并确定每个 KV 键值对的来源。

message StoreKVPair {
  optional string store_key = 1; // 该键值对来源的 KVStore 的 store key
  required bool set = 2; // true 表示 set 操作，false 表示 delete 操作
  required bytes key = 3;
  required bytes value = 4;
}

ListenKVStore

我们将创建一个新的 Store 类型 listenkv.Store，由 rootmulti store 用来包装 KVStore，以启用状态监听。我们将使用 MemoryListener 来配置这个 Store，它会收集状态变化并输出到特定目标。

// Store 实现了启用监听能力的 KVStore 接口。
// 会在每次核心 KVStore 调用时追踪操作，并以正确的键和操作权限
// 写入任意底层监听器
type Store struct {
    parent    types.KVStore
    listener  *types.MemoryListener
    parentStoreKey types.StoreKey
}

// NewStore 根据父级 KVStore 实现和一个缓冲写入器，
// 返回一个新的 traceKVStore 引用。
func NewStore(parent types.KVStore, psk types.StoreKey, listener *types.MemoryListener) *Store {
    return &Store{
    parent: parent, listener: listener, parentStoreKey: psk
}
}

// Set 实现 KVStore 接口。它会追踪一次写操作，
// 并将 Set 调用委托给父级 KVStore。
func (s *Store)

Set(key []byte, value []byte) {
    types.AssertValidKey(key)

s.parent.Set(key, value)

s.listener.OnWrite(s.parentStoreKey, key, value, false)
}

// Delete 实现 KVStore 接口。它会追踪一次写操作，
// 并将 Delete 调用委托给父级 KVStore。
func (s *Store)

Delete(key []byte) {
    s.parent.Delete(key)

s.listener.OnWrite(s.parentStoreKey, key, nil, true)
}

MultiStore 接口更新

我们将更新 CommitMultiStore 接口，以便将 Memorylistener 包装到特定的 KVStore 上。请注意，MemoryListener 将由具体的 rootmulti 实现在内部附加。

type CommitMultiStore interface {
    ...

    // AddListeners 为属于所提供 StoreKey 的 KVStore 添加监听器
    AddListeners(keys []StoreKey)

    // PopStateCache 返回 MemoryListener 中累积的状态变化消息
    PopStateCache() []StoreKVPair
}

MultiStore 实现更新

我们将调整 rootmulti 的 GetKVStore 方法：如果为该 Store 启用了监听，则用 listenkv.Store 包装返回的 KVStore。

func (rs *Store)

GetKVStore(key types.StoreKey)

types.KVStore {
    store := rs.stores[key].(types.KVStore)
    if rs.TracingEnabled() {
    store = tracekv.NewStore(store, rs.traceWriter, rs.traceContext)
}
    if rs.ListeningEnabled(key) {
    store = listenkv.NewStore(store, key, rs.listeners[key])
}

return store
}

我们将实现 AddListeners，在内部管理 KVStore 监听器，并实现 PopStateCache 以提供获取当前状态的方式。

// AddListeners 为特定 KVStore 添加状态变化监听器
func (rs *Store)

AddListeners(keys []types.StoreKey) {
    listener := types.NewMemoryListener()
    for i := range keys {
    rs.listeners[keys[i]] = listener
}
}

func (rs *Store)

PopStateCache() []types.StoreKVPair {
    var cache []types.StoreKVPair
    for _, ls := range rs.listeners {
    cache = append(cache, ls.PopStateCache()...)
}

sort.SliceStable(cache, func(i, j int)

bool {
    return cache[i].StoreKey < cache[j].StoreKey
})

return cache
}

我们还将调整 rootmulti 的 CacheMultiStore 和 CacheMultiStoreWithVersion 方法，以在缓存层启用监听。

func (rs *Store)

CacheMultiStore()

types.CacheMultiStore {
    stores := make(map[types.StoreKey]types.CacheWrapper)
    for k, v := range rs.stores {
    store := v.(types.KVStore)
        // 接入 listenkv.Store，以便监听器能够观察来自缓存存储的写入，
        // 在缓存存储上设置相同的监听器会观察到重复写入。
    if rs.ListeningEnabled(k) {
    store = listenkv.NewStore(store, k, rs.listeners[k])
}

stores[k] = store
}

return cachemulti.NewStore(rs.db, stores, rs.keysByName, rs.traceWriter, rs.getTracingContext())
}

func (rs *Store)

CacheMultiStoreWithVersion(version int64) (types.CacheMultiStore, error) {
 // ...

        // 接入 listenkv.Store，以便监听器能够观察来自缓存存储的写入，
        // 在缓存存储上设置相同的监听器会观察到重复写入。
    if rs.ListeningEnabled(key) {
    cacheStore = listenkv.NewStore(cacheStore, key, rs.listeners[key])
}

cachedStores[key] = cacheStore
}

return cachemulti.NewStore(rs.db, cachedStores, rs.keysByName, rs.traceWriter, rs.getTracingContext()), nil
}

暴露数据

流式服务

我们将引入一个新的 ABCIListener 接口，把它接入 BaseApp，并转发 ABCI 请求与响应，以便服务能够将状态变化与 ABCI 请求分组关联起来。

// baseapp/streaming.go

// ABCIListener 是我们对外暴露为流式服务的接口。
type ABCIListener interface {
	// ListenFinalizeBlock 使用最新的 FinalizeBlock 消息更新流式服务
	ListenFinalizeBlock(ctx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock)

error
	// ListenCommit 使用最新的 Commit 消息和状态变化更新流式服务
	ListenCommit(ctx context.Context, res abci.ResponseCommit, changeSet []*StoreKVPair)

error
}

BaseApp 注册

我们将向 BaseApp 添加一个新方法，以支持注册 StreamingService：

// SetStreamingService 用于将流式服务设置到 BaseApp 钩子中，并把监听器加载到 multistore
func (app *BaseApp)

SetStreamingService(s ABCIListener) {
   // 在 BaseApp 中注册 StreamingService
   // BaseApp 会将 BeginBlock、DeliverTx 和 EndBlock 的请求与响应传递给流式服务，
   // 以更新它们的 ABCI 上下文
   app.abciListeners = append(app.abciListeners, s)
}

我们还将向 BaseApp 结构体添加两个新字段：

type BaseApp struct {

    ...

    // abciListenersAsync 用于决定 abciListeners 是否异步运行。
    // 当 abciListenersAsync=false 且 stopNodeOnABCIListenerErr=false 时，监听器将同步运行，但不会停止节点。
    // 当 abciListenersAsync=true 时，stopNodeOnABCIListenerErr 将被忽略。
    abciListenersAsync bool

    // stopNodeOnABCIListenerErr 在 ABCI 流式服务监听产生错误时停止节点。
    // stopNodeOnABCIListenerErr=true 必须与 abciListenersAsync=false 配合使用。
    stopNodeOnABCIListenerErr bool
}

ABCI 事件钩子

我们将修改 FinalizeBlock 和 Commit 方法，以便将 ABCI 请求和响应传递给任何已向 BaseApp 注册的流式服务钩子。

func (app *BaseApp)

FinalizeBlock(req abci.RequestFinalizeBlock)

abci.ResponseFinalizeBlock {
    var abciRes abci.ResponseFinalizeBlock
    defer func() {
        // call the streaming service hook with the FinalizeBlock messages
    for _, abciListener := range app.abciListeners {
    ctx := app.finalizeState.ctx
    blockHeight := ctx.BlockHeight()
    if app.abciListenersAsync {
    go func(req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock) {
    if err := app.abciListener.FinalizeBlock(blockHeight, req, res); err != nil {
    app.logger.Error("FinalizeBlock listening hook failed", "height", blockHeight, "err", err)
}
 
}(req, abciRes)
}

else {
    if err := app.abciListener.ListenFinalizeBlock(blockHeight, req, res); err != nil {
    app.logger.Error("FinalizeBlock listening hook failed", "height", blockHeight, "err", err)
    if app.stopNodeOnABCIListenerErr {
    os.Exit(1)
}
 
}
 
}
 
}
 
}()

    ...

    return abciRes
}

func (app *BaseApp)

Commit()

abci.ResponseCommit {

    ...
    res := abci.ResponseCommit{
    Data:         commitID.Hash,
    RetainHeight: retainHeight,
}

    // call the streaming service hook with the Commit messages
    for _, abciListener := range app.abciListeners {
    ctx := app.deliverState.ctx
    blockHeight := ctx.BlockHeight()
    changeSet := app.cms.PopStateCache()
    if app.abciListenersAsync {
    go func(res abci.ResponseCommit, changeSet []store.StoreKVPair) {
    if err := app.abciListener.ListenCommit(ctx, res, changeSet); err != nil {
    app.logger.Error("ListenCommit listening hook failed", "height", blockHeight, "err", err)
}
 
}(res, changeSet)
}

else {
    if err := app.abciListener.ListenCommit(ctx, res, changeSet); err != nil {
    app.logger.Error("ListenCommit listening hook failed", "height", blockHeight, "err", err)
    if app.stopNodeOnABCIListenerErr {
    os.Exit(1)
}
 
}
 
}
 
}

    ...

    return res
}

Go 插件系统

我们提出一种插件架构，用于加载和运行 Streaming 插件及其他类型的实现。我们将引入一个基于 gRPC 的插件系统，用于加载和运行 Cosmos-SDK 插件。该插件系统使用 hashicorp/go-plugin。每个插件都必须有一个结构体来实现 plugin.Plugin 接口，以及一个 Impl 接口来处理通过 gRPC 传递的消息。每个插件还必须为 gRPC 服务定义消息协议：

// streaming/plugins/abci/{
    plugin_version
}/interface.go

// Handshake is a common handshake that is shared by streaming and host.
// This prevents users from executing bad plugins or executing a plugin
// directory. It is a UX feature, not a security feature.
var Handshake = plugin.HandshakeConfig{
    ProtocolVersion:  1,
    MagicCookieKey:   "ABCI_LISTENER_PLUGIN",
    MagicCookieValue: "ef78114d-7bdf-411c-868f-347c99a78345",
}

// ListenerPlugin is the base struc for all kinds of go-plugin implementations
// It will be included in interfaces of different Plugins
type ABCIListenerPlugin struct {
    // GRPCPlugin must still implement the Plugin interface
    plugin.Plugin
    // Concrete implementation, written in Go. This is only used for plugins
    // that are written in Go.
    Impl baseapp.ABCIListener
}

func (p *ListenerGRPCPlugin)

GRPCServer(_ *plugin.GRPCBroker, s *grpc.Server)

error {
    RegisterABCIListenerServiceServer(s, &GRPCServer{
    Impl: p.Impl
})

return nil
}

func (p *ListenerGRPCPlugin)

GRPCClient(
    _ context.Context,
    _ *plugin.GRPCBroker,
    c *grpc.ClientConn,
) (interface{
}, error) {
    return &GRPCClient{
    client: NewABCIListenerServiceClient(c)
}, nil
}

plugin.Plugin 接口有两个方法：Client 和 Server。对于我们的 gRPC 服务，这两个方法分别是 GRPCClient 和 GRPCServer。 Impl 字段保存了用 Go 编写的 baseapp.ABCIListener 接口的具体实现。注意：这仅用于以 Go 编写的插件实现。这种插件系统的优势在于，插件作者可以在每个插件内部以适合其用例的方式定义消息协议。例如，当需要监听状态变更时，可以如下定义 ABCIListener 消息协议（仅用于说明）。当不需要监听状态变更时，可以从协议中省略 ListenCommit。

syntax = "proto3";

...

message Empty {}

message ListenFinalizeBlockRequest {
  RequestFinalizeBlock  req = 1;
  ResponseFinalizeBlock res = 2;
}
message ListenCommitRequest {
  int64                block_height = 1;
  ResponseCommit       res          = 2;
  repeated StoreKVPair changeSet    = 3;
}

// plugin that listens to state changes
service ABCIListenerService {
  rpc ListenFinalizeBlock(ListenFinalizeBlockRequest) returns (Empty);
  rpc ListenCommit(ListenCommitRequest) returns (Empty);
}

...
// plugin that doesn't listen to state changes
service ABCIListenerService {
  rpc ListenFinalizeBlock(ListenFinalizeBlockRequest) returns (Empty);
  rpc ListenCommit(ListenCommitRequest) returns (Empty);
}

上述服务的实现：

// streaming/plugins/abci/{
    plugin_version
}/grpc.go

var (
    _ baseapp.ABCIListener = (*GRPCClient)(nil)
)

// GRPCClient is an implementation of the ABCIListener and ABCIListenerPlugin interfaces that talks over RPC.
type GRPCClient struct {
    client ABCIListenerServiceClient
}

func (m *GRPCClient)

ListenFinalizeBlock(goCtx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock)

error {
    ctx := sdk.UnwrapSDKContext(goCtx)
    _, err := m.client.ListenDeliverTx(ctx, &ListenDeliverTxRequest{
    BlockHeight: ctx.BlockHeight(),
    Req: req,
    Res: res
})

return err
}

func (m *GRPCClient)

ListenCommit(goCtx context.Context, res abci.ResponseCommit, changeSet []store.StoreKVPair)

error {
    ctx := sdk.UnwrapSDKContext(goCtx)
    _, err := m.client.ListenCommit(ctx, &ListenCommitRequest{
    BlockHeight: ctx.BlockHeight(),
    Res: res,
    ChangeSet: changeSet
})

return err
}

// GRPCServer is the gRPC server that GRPCClient talks to.
type GRPCServer struct {
    // This is the real implementation
    Impl baseapp.ABCIListener
}

func (m *GRPCServer)

ListenFinalizeBlock(ctx context.Context, req *ListenFinalizeBlockRequest) (*Empty, error) {
    return &Empty{
}, m.Impl.ListenFinalizeBlock(ctx, req.Req, req.Res)
}

func (m *GRPCServer)

ListenCommit(ctx context.Context, req *ListenCommitRequest) (*Empty, error) {
    return &Empty{
}, m.Impl.ListenCommit(ctx, req.Res, req.ChangeSet)
}

以及预编译的 Go 插件 Impl（这仅用于以 Go 编写的插件）：

// streaming/plugins/abci/{
    plugin_version
}/impl/plugin.go

// Plugins are pre-compiled and loaded by the plugin system

// ABCIListener is the implementation of the baseapp.ABCIListener interface
type ABCIListener struct{
}

func (m *ABCIListenerPlugin)

ListenFinalizeBlock(ctx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock)

error {
    // send data to external system
}

func (m *ABCIListenerPlugin)

ListenCommit(ctx context.Context, res abci.ResponseCommit, changeSet []store.StoreKVPair)

error {
    // send data to external system
}

func main() {
    plugin.Serve(&plugin.ServeConfig{
    HandshakeConfig: grpc_abci_v1.Handshake,
    Plugins: map[string]plugin.Plugin{
           "grpc_plugin_v1": &grpc_abci_v1.ABCIListenerGRPCPlugin{
    Impl: &ABCIListenerPlugin{
}},
},

        // A non-nil value here enables gRPC serving for this streaming...
        GRPCServer: plugin.DefaultGRPCServer,
})
}

我们将引入一个插件加载系统，它将返回 (interface{}, error)。这带来了使用带版本插件的优势，即插件接口和 gRPC 协议可以随着时间演进而变化。此外，它还支持构建独立插件，以便通过 gRPC 暴露系统的不同部分。

func NewStreamingPlugin(name string, logLevel string) (interface{
}, error) {
    logger := hclog.New(&hclog.LoggerOptions{
    Output: hclog.DefaultOutput,
    Level:  toHclogLevel(logLevel),
    Name:   fmt.Sprintf("plugin.%s", name),
})

    // We're a host. Start by launching the streaming process.
    env := os.Getenv(GetPluginEnvKey(name))
    client := plugin.NewClient(&plugin.ClientConfig{
    HandshakeConfig: HandshakeMap[name],
    Plugins:         PluginMap,
    Cmd:             exec.Command("sh", "-c", env),
    Logger:          logger,
    AllowedProtocols: []plugin.Protocol{
    plugin.ProtocolNetRPC, plugin.ProtocolGRPC
},
})

    // Connect via RPC
    rpcClient, err := client.Client()
    if err != nil {
    return nil, err
}

    // Request streaming plugin
    return rpcClient.Dispense(name)
}

我们为应用提出一个 RegisterStreamingPlugin 函数，用于向应用的 BaseApp 注册 NewStreamingPlugin。 Streaming 插件可以是 Any 类型；因此，该函数接收的是接口而不是具体类型。例如，我们可以有 ABCIListener、WasmListener 或 IBCListener 类型的插件。注意，RegisterStreamingPluing 函数只是辅助函数，并非必需。插件注册很容易从应用侧直接移到 BaseApp 中。

// baseapp/streaming.go

// RegisterStreamingPlugin registers streaming plugins with the App.
// This method returns an error if a plugin is not supported.
func RegisterStreamingPlugin(
    bApp *BaseApp,
    appOpts servertypes.AppOptions,
    keys map[string]*types.KVStoreKey,
    streamingPlugin interface{
},
)

error {
    switch t := streamingPlugin.(type) {
    case ABCIListener:
        registerABCIListenerPlugin(bApp, appOpts, keys, t)

default:
        return fmt.Errorf("unexpected plugin type %T", t)
}

return nil
}

func registerABCIListenerPlugin(
    bApp *BaseApp,
    appOpts servertypes.AppOptions,
    keys map[string]*store.KVStoreKey,
    abciListener ABCIListener,
) {
    asyncKey := fmt.Sprintf("%s.%s.%s", StreamingTomlKey, StreamingABCITomlKey, StreamingABCIAsync)
    async := cast.ToBool(appOpts.Get(asyncKey))
    stopNodeOnErrKey := fmt.Sprintf("%s.%s.%s", StreamingTomlKey, StreamingABCITomlKey, StreamingABCIStopNodeOnErrTomlKey)
    stopNodeOnErr := cast.ToBool(appOpts.Get(stopNodeOnErrKey))
    keysKey := fmt.Sprintf("%s.%s.%s", StreamingTomlKey, StreamingABCITomlKey, StreamingABCIKeysTomlKey)
    exposeKeysStr := cast.ToStringSlice(appOpts.Get(keysKey))
    exposedKeys := exposeStoreKeysSorted(exposeKeysStr, keys)

bApp.cms.AddListeners(exposedKeys)

app.SetStreamingManager(
		storetypes.StreamingManager{
    ABCIListeners: []storetypes.ABCIListener{
    abciListener
},
    StopNodeOnErr: stopNodeOnErr,
},
	)
}

func exposeAll(list []string)

bool {
    for _, ele := range list {
    if ele == "*" {
    return true
}
 
}

return false
}

func exposeStoreKeys(keysStr []string, keys map[string]*types.KVStoreKey) []types.StoreKey {
    var exposeStoreKeys []types.StoreKey
    if exposeAll(keysStr) {
    exposeStoreKeys = make([]types.StoreKey, 0, len(keys))
    for _, storeKey := range keys {
    exposeStoreKeys = append(exposeStoreKeys, storeKey)
}
 
}

else {
    exposeStoreKeys = make([]types.StoreKey, 0, len(keysStr))
    for _, keyStr := range keysStr {
    if storeKey, ok := keys[keyStr]; ok {
    exposeStoreKeys = append(exposeStoreKeys, storeKey)
}
 
}
 
}
    // sort storeKeys for deterministic output
    sort.SliceStable(exposeStoreKeys, func(i, j int)

bool {
    return exposeStoreKeys[i].Name() < exposeStoreKeys[j].Name()
})

return exposeStoreKeys
}

NewStreamingPlugin 和 RegisterStreamingPlugin 函数用于将插件注册到应用的 BaseApp。例如，在 NewSimApp 中：

func NewSimApp(
    logger log.Logger,
    db dbm.DB,
    traceStore io.Writer,
    loadLatest bool,
    appOpts servertypes.AppOptions,
    baseAppOptions ...func(*baseapp.BaseApp),
) *SimApp {

    ...
    keys := sdk.NewKVStoreKeys(
       authtypes.StoreKey, banktypes.StoreKey, stakingtypes.StoreKey,
       minttypes.StoreKey, distrtypes.StoreKey, slashingtypes.StoreKey,
       govtypes.StoreKey, paramstypes.StoreKey, ibchost.StoreKey, upgradetypes.StoreKey,
       evidencetypes.StoreKey, ibctransfertypes.StoreKey, capabilitytypes.StoreKey,
    )

    ...

    // register streaming services
    streamingCfg := cast.ToStringMap(appOpts.Get(baseapp.StreamingTomlKey))
    for service := range streamingCfg {
    pluginKey := fmt.Sprintf("%s.%s.%s", baseapp.StreamingTomlKey, service, baseapp.StreamingPluginTomlKey)
    pluginName := strings.TrimSpace(cast.ToString(appOpts.Get(pluginKey)))
    if len(pluginName) > 0 {
    logLevel := cast.ToString(appOpts.Get(flags.FlagLogLevel))

plugin, err := streaming.NewStreamingPlugin(pluginName, logLevel)
    if err != nil {
    tmos.Exit(err.Error())
}
    if err := baseapp.RegisterStreamingPlugin(bApp, appOpts, keys, plugin); err != nil {
    tmos.Exit(err.Error())
}
 
}
 
}

return app

配置

插件系统将在应用的 TOML 配置文件中进行配置。

# gRPC streaming
[streaming]

# ABCI streaming service
[streaming.abci]

# The plugin version to use for ABCI listening
plugin = "abci_v1"

# List of kv store keys to listen to for state changes.

# Set to ["*"] to expose all keys.
keys = ["*"]

# Enable abciListeners to run asynchronously.

# When abciListenersAsync=false and stopNodeOnABCIListenerErr=false listeners will run synchronized but will not stop the node.

# When abciListenersAsync=true stopNodeOnABCIListenerErr will be ignored.
async = false

Whether to stop the node on message deliver error.

stop-node-on-err = true

`ABCIListener` 插件将有四个配置参数：`streaming.abci.plugin`、`streaming.abci.keys`、`streaming.abci.async` 和 `streaming.abci.stop-node-on-err`。
`streaming.abci.plugin` 是我们希望用于流式处理的插件名称，`streaming.abci.keys` 是它要监听的 store key 集合，
`streaming.abci.async` 是一个用于启用异步监听的布尔值，而 `streaming.abci.stop-node-on-err` 也是一个布尔值：当其为 true，且运行在同步模式
`streaming.abci.async=false` 时，会停止节点。注意，如果 `streaming.abci.async=true`，则 `streaming.abci.stop-node-on-err=true` 会被忽略。

上述配置通过将插件添加到 `[streaming]` 配置节中，
并使用 `RegisterStreamingPlugin` 辅助函数注册该插件，从而支持额外的流式插件。

注意，每个插件都必须包含 `streaming.{service}.plugin` 属性，因为这是在 App 中执行插件查找和注册的必要条件。
其他所有属性则由各个服务自行定义。


#### 流的编码与解码

ADR-038 引入了接口和类型，用于从 KVStore 中流式输出状态变更，将这些
数据与其相关的 ABCI 请求和响应关联起来，并注册一个服务来消费这些数据，并以最终格式将其流式传输到某个目标。
此 ADR 并未规定最终数据格式，而是将该格式的定义和文档编写交由具体插件实现。
我们采用这种方式，是因为要支持广泛的流式服务插件，就必须保留最终格式的灵活性。例如，
将数据写入一组文件的流式服务，其数据格式会不同于写入 Kafka topic 的数据格式。


## 影响

这些变更将提供一种实时订阅 KVStore 状态变更的方式。


### 向后兼容性

* 此 ADR 修改了 `CommitMultiStore` 接口，支持该接口旧版本的实现将无法支持新版本


### 正面影响

* 能够实时监听 KVStore 状态变更，并将这些事件暴露给外部消费者


### 负面影响

* 修改了 `CommitMultiStore` 接口及其实现


### 中性影响

* 为 cosmos 应用的配置和运行引入了额外但可选的复杂性
* 如果应用开发者选择使用这些特性来暴露数据，他们需要了解此类数据暴露结合其具体应用场景所带来的影响和风险

---

## Changelog

* 11/23/2020: Initial draft
* 10/06/2022: Introduce plugin system based on hashicorp/go-plugin
* 10/14/2022:
  * Add `ListenCommit`, flatten the state writes in a block to a single batch.
  * Remove listeners from cache stores, should only listen to `rootmulti.Store`.
  * Remove `HaltAppOnDeliveryError()`, the errors are propagated by default, the implementations should return nil if don't want to propogate errors.
* 26/05/2023: Update with ABCI 2.0

## Status

Proposed

## Abstract

This ADR defines a set of changes to enable listening to state changes of individual KVStores and exposing these data to consumers.

## Context

Currently, KVStore data can be remotely accessed through [Queries](https://github.com/cosmos/cosmos-sdk/blob/master/docs/building-modules/messages-and-queries.md#queries)
which proceed either through Tendermint and the ABCI, or through the gRPC server.
In addition to these request/response queries, it would be beneficial to have a means of listening to state changes as they occur in real time.

## Decision

We will modify the `CommitMultiStore` interface and its concrete (`rootmulti`) implementations and introduce a new `listenkv.Store` to allow listening to state changes in underlying KVStores. We don't need to listen to cache stores, because we can't be sure that the writes will be committed eventually, and the writes are duplicated in `rootmulti.Store` eventually, so we should only listen to `rootmulti.Store`.
We will introduce a plugin system for configuring and running streaming services that write these state changes and their surrounding ABCI message context to different destinations.

### Listening

In a new file, `store/types/listening.go`, we will create a `MemoryListener` struct for streaming out protobuf encoded KV pairs state changes from a KVStore.
The `MemoryListener` will be used internally by the concrete `rootmulti` implementation to collect state changes from KVStores.

```go expandable
// MemoryListener listens to the state writes and accumulate the records in memory.
type MemoryListener struct {
    stateCache []StoreKVPair
}

// NewMemoryListener creates a listener that accumulate the state writes in memory.
func NewMemoryListener() *MemoryListener {
    return &MemoryListener{
}
}

// OnWrite writes state change events to the internal cache
func (fl *MemoryListener)

OnWrite(storeKey StoreKey, key []byte, value []byte, delete bool) {
    fl.stateCache = append(fl.stateCache, StoreKVPair{
    StoreKey: storeKey.Name(),
    Delete:   delete,
    Key:      key,
    Value:    value,
})
}

// PopStateCache returns the current state caches and set to nil
func (fl *MemoryListener)

PopStateCache() []StoreKVPair {
    res := fl.stateCache
	fl.stateCache = nil
	return res
}

We will also define a protobuf type for the KV pairs. In addition to the key and value fields this message will include the StoreKey for the originating KVStore so that we can collect information from separate KVStores and determine the source of each KV pair.

message StoreKVPair {
  optional string store_key = 1; // the store key for the KVStore this pair originates from
  required bool set = 2; // true indicates a set operation, false indicates a delete operation
  required bytes key = 3;
  required bytes value = 4;
}

ListenKVStore

We will create a new Store type listenkv.Store that the rootmulti store will use to wrap a KVStore to enable state listening. We will configure the Store with a MemoryListener which will collect state changes for output to specific destinations.

// Store implements the KVStore interface with listening enabled.
// Operations are traced on each core KVStore call and written to any of the
// underlying listeners with the proper key and operation permissions
type Store struct {
    parent    types.KVStore
    listener  *types.MemoryListener
    parentStoreKey types.StoreKey
}

// NewStore returns a reference to a new traceKVStore given a parent
// KVStore implementation and a buffered writer.
func NewStore(parent types.KVStore, psk types.StoreKey, listener *types.MemoryListener) *Store {
    return &Store{
    parent: parent, listener: listener, parentStoreKey: psk
}
}

// Set implements the KVStore interface. It traces a write operation and
// delegates the Set call to the parent KVStore.
func (s *Store)

Set(key []byte, value []byte) {
    types.AssertValidKey(key)

s.parent.Set(key, value)

s.listener.OnWrite(s.parentStoreKey, key, value, false)
}

// Delete implements the KVStore interface. It traces a write operation and
// delegates the Delete call to the parent KVStore.
func (s *Store)

Delete(key []byte) {
    s.parent.Delete(key)

s.listener.OnWrite(s.parentStoreKey, key, nil, true)
}

MultiStore interface updates

We will update the CommitMultiStore interface to allow us to wrap a Memorylistener to a specific KVStore. Note that the MemoryListener will be attached internally by the concrete rootmulti implementation.

type CommitMultiStore interface {
    ...

    // AddListeners adds a listener for the KVStore belonging to the provided StoreKey
    AddListeners(keys []StoreKey)

    // PopStateCache returns the accumulated state change messages from MemoryListener
    PopStateCache() []StoreKVPair
}

MultiStore implementation updates

We will adjust the rootmulti GetKVStore method to wrap the returned KVStore with a listenkv.Store if listening is turned on for that Store.

func (rs *Store)

GetKVStore(key types.StoreKey)

types.KVStore {
    store := rs.stores[key].(types.KVStore)
    if rs.TracingEnabled() {
    store = tracekv.NewStore(store, rs.traceWriter, rs.traceContext)
}
    if rs.ListeningEnabled(key) {
    store = listenkv.NewStore(store, key, rs.listeners[key])
}

return store
}

We will implement AddListeners to manage KVStore listeners internally and implement PopStateCache for a means of retrieving the current state.

// AddListeners adds state change listener for a specific KVStore
func (rs *Store)

AddListeners(keys []types.StoreKey) {
    listener := types.NewMemoryListener()
    for i := range keys {
    rs.listeners[keys[i]] = listener
}
}

func (rs *Store)

PopStateCache() []types.StoreKVPair {
    var cache []types.StoreKVPair
    for _, ls := range rs.listeners {
    cache = append(cache, ls.PopStateCache()...)
}

sort.SliceStable(cache, func(i, j int)

bool {
    return cache[i].StoreKey < cache[j].StoreKey
})

return cache
}

We will also adjust the rootmulti CacheMultiStore and CacheMultiStoreWithVersion methods to enable listening in the cache layer.

func (rs *Store)

CacheMultiStore()

types.CacheMultiStore {
    stores := make(map[types.StoreKey]types.CacheWrapper)
    for k, v := range rs.stores {
    store := v.(types.KVStore)
        // Wire the listenkv.Store to allow listeners to observe the writes from the cache store,
        // set same listeners on cache store will observe duplicated writes.
    if rs.ListeningEnabled(k) {
    store = listenkv.NewStore(store, k, rs.listeners[k])
}

stores[k] = store
}

return cachemulti.NewStore(rs.db, stores, rs.keysByName, rs.traceWriter, rs.getTracingContext())
}

func (rs *Store)

CacheMultiStoreWithVersion(version int64) (types.CacheMultiStore, error) {
 // ...

        // Wire the listenkv.Store to allow listeners to observe the writes from the cache store,
        // set same listeners on cache store will observe duplicated writes.
    if rs.ListeningEnabled(key) {
    cacheStore = listenkv.NewStore(cacheStore, key, rs.listeners[key])
}

cachedStores[key] = cacheStore
}

return cachemulti.NewStore(rs.db, cachedStores, rs.keysByName, rs.traceWriter, rs.getTracingContext()), nil
}

Exposing the data

Streaming Service

We will introduce a new ABCIListener interface that plugs into the BaseApp and relays ABCI requests and responses so that the service can group the state changes with the ABCI requests.

// baseapp/streaming.go

// ABCIListener is the interface that we're exposing as a streaming service.
type ABCIListener interface {
	// ListenFinalizeBlock updates the streaming service with the latest FinalizeBlock messages
	ListenFinalizeBlock(ctx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock)

error
	// ListenCommit updates the steaming service with the latest Commit messages and state changes
	ListenCommit(ctx context.Context, res abci.ResponseCommit, changeSet []*StoreKVPair)

error
}

BaseApp Registration

We will add a new method to the BaseApp to enable the registration of StreamingServices:

// SetStreamingService is used to set a streaming service into the BaseApp hooks and load the listeners into the multistore
func (app *BaseApp)

SetStreamingService(s ABCIListener) {
   // register the StreamingService within the BaseApp
   // BaseApp will pass BeginBlock, DeliverTx, and EndBlock requests and responses to the streaming services to update their ABCI context
   app.abciListeners = append(app.abciListeners, s)
}

We will add two new fields to the BaseApp struct:

type BaseApp struct {

    ...

    // abciListenersAsync for determining if abciListeners will run asynchronously.
    // When abciListenersAsync=false and stopNodeOnABCIListenerErr=false listeners will run synchronized but will not stop the node.
    // When abciListenersAsync=true stopNodeOnABCIListenerErr will be ignored.
    abciListenersAsync bool

    // stopNodeOnABCIListenerErr halts the node when ABCI streaming service listening results in an error.
    // stopNodeOnABCIListenerErr=true must be paired with abciListenersAsync=false.
    stopNodeOnABCIListenerErr bool
}

ABCI Event Hooks

We will modify the FinalizeBlock and Commit methods to pass ABCI requests and responses to any streaming service hooks registered with the BaseApp.

func (app *BaseApp)

FinalizeBlock(req abci.RequestFinalizeBlock)

abci.ResponseFinalizeBlock {
    var abciRes abci.ResponseFinalizeBlock
    defer func() {
        // call the streaming service hook with the FinalizeBlock messages
    for _, abciListener := range app.abciListeners {
    ctx := app.finalizeState.ctx
    blockHeight := ctx.BlockHeight()
    if app.abciListenersAsync {
    go func(req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock) {
    if err := app.abciListener.FinalizeBlock(blockHeight, req, res); err != nil {
    app.logger.Error("FinalizeBlock listening hook failed", "height", blockHeight, "err", err)
}
 
}(req, abciRes)
}

else {
    if err := app.abciListener.ListenFinalizeBlock(blockHeight, req, res); err != nil {
    app.logger.Error("FinalizeBlock listening hook failed", "height", blockHeight, "err", err)
    if app.stopNodeOnABCIListenerErr {
    os.Exit(1)
}
 
}
 
}
 
}
 
}()

    ...

    return abciRes
}

func (app *BaseApp)

Commit()

abci.ResponseCommit {

    ...
    res := abci.ResponseCommit{
    Data:         commitID.Hash,
    RetainHeight: retainHeight,
}

    // call the streaming service hook with the Commit messages
    for _, abciListener := range app.abciListeners {
    ctx := app.deliverState.ctx
    blockHeight := ctx.BlockHeight()
    changeSet := app.cms.PopStateCache()
    if app.abciListenersAsync {
    go func(res abci.ResponseCommit, changeSet []store.StoreKVPair) {
    if err := app.abciListener.ListenCommit(ctx, res, changeSet); err != nil {
    app.logger.Error("ListenCommit listening hook failed", "height", blockHeight, "err", err)
}
 
}(res, changeSet)
}

else {
    if err := app.abciListener.ListenCommit(ctx, res, changeSet); err != nil {
    app.logger.Error("ListenCommit listening hook failed", "height", blockHeight, "err", err)
    if app.stopNodeOnABCIListenerErr {
    os.Exit(1)
}
 
}
 
}
 
}

    ...

    return res
}

Go Plugin System

We propose a plugin architecture to load and run Streaming plugins and other types of implementations. We will introduce a plugin system over gRPC that is used to load and run Cosmos-SDK plugins. The plugin system uses hashicorp/go-plugin. Each plugin must have a struct that implements the plugin.Plugin interface and an Impl interface for processing messages over gRPC. Each plugin must also have a message protocol defined for the gRPC service:

// streaming/plugins/abci/{
    plugin_version
}/interface.go

// Handshake is a common handshake that is shared by streaming and host.
// This prevents users from executing bad plugins or executing a plugin
// directory. It is a UX feature, not a security feature.
var Handshake = plugin.HandshakeConfig{
    ProtocolVersion:  1,
    MagicCookieKey:   "ABCI_LISTENER_PLUGIN",
    MagicCookieValue: "ef78114d-7bdf-411c-868f-347c99a78345",
}

// ListenerPlugin is the base struc for all kinds of go-plugin implementations
// It will be included in interfaces of different Plugins
type ABCIListenerPlugin struct {
    // GRPCPlugin must still implement the Plugin interface
    plugin.Plugin
    // Concrete implementation, written in Go. This is only used for plugins
    // that are written in Go.
    Impl baseapp.ABCIListener
}

func (p *ListenerGRPCPlugin)

GRPCServer(_ *plugin.GRPCBroker, s *grpc.Server)

error {
    RegisterABCIListenerServiceServer(s, &GRPCServer{
    Impl: p.Impl
})

return nil
}

func (p *ListenerGRPCPlugin)

GRPCClient(
    _ context.Context,
    _ *plugin.GRPCBroker,
    c *grpc.ClientConn,
) (interface{
}, error) {
    return &GRPCClient{
    client: NewABCIListenerServiceClient(c)
}, nil
}

The plugin.Plugin interface has two methods Client and Server. For our GRPC service these are GRPCClient and GRPCServer The Impl field holds the concrete implementation of our baseapp.ABCIListener interface written in Go. Note: this is only used for plugin implementations written in Go. The advantage of having such a plugin system is that within each plugin authors can define the message protocol in a way that fits their use case. For example, when state change listening is desired, the ABCIListener message protocol can be defined as below (for illustrative purposes only). When state change listening is not desired than ListenCommit can be omitted from the protocol.

syntax = "proto3";

...

message Empty {}

message ListenFinalizeBlockRequest {
  RequestFinalizeBlock  req = 1;
  ResponseFinalizeBlock res = 2;
}
message ListenCommitRequest {
  int64                block_height = 1;
  ResponseCommit       res          = 2;
  repeated StoreKVPair changeSet    = 3;
}

// plugin that listens to state changes
service ABCIListenerService {
  rpc ListenFinalizeBlock(ListenFinalizeBlockRequest) returns (Empty);
  rpc ListenCommit(ListenCommitRequest) returns (Empty);
}

...
// plugin that doesn't listen to state changes
service ABCIListenerService {
  rpc ListenFinalizeBlock(ListenFinalizeBlockRequest) returns (Empty);
  rpc ListenCommit(ListenCommitRequest) returns (Empty);
}

Implementing the service above:

// streaming/plugins/abci/{
    plugin_version
}/grpc.go

var (
    _ baseapp.ABCIListener = (*GRPCClient)(nil)
)

// GRPCClient is an implementation of the ABCIListener and ABCIListenerPlugin interfaces that talks over RPC.
type GRPCClient struct {
    client ABCIListenerServiceClient
}

func (m *GRPCClient)

ListenFinalizeBlock(goCtx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock)

error {
    ctx := sdk.UnwrapSDKContext(goCtx)
    _, err := m.client.ListenDeliverTx(ctx, &ListenDeliverTxRequest{
    BlockHeight: ctx.BlockHeight(),
    Req: req,
    Res: res
})

return err
}

func (m *GRPCClient)

ListenCommit(goCtx context.Context, res abci.ResponseCommit, changeSet []store.StoreKVPair)

error {
    ctx := sdk.UnwrapSDKContext(goCtx)
    _, err := m.client.ListenCommit(ctx, &ListenCommitRequest{
    BlockHeight: ctx.BlockHeight(),
    Res: res,
    ChangeSet: changeSet
})

return err
}

// GRPCServer is the gRPC server that GRPCClient talks to.
type GRPCServer struct {
    // This is the real implementation
    Impl baseapp.ABCIListener
}

func (m *GRPCServer)

ListenFinalizeBlock(ctx context.Context, req *ListenFinalizeBlockRequest) (*Empty, error) {
    return &Empty{
}, m.Impl.ListenFinalizeBlock(ctx, req.Req, req.Res)
}

func (m *GRPCServer)

ListenCommit(ctx context.Context, req *ListenCommitRequest) (*Empty, error) {
    return &Empty{
}, m.Impl.ListenCommit(ctx, req.Res, req.ChangeSet)
}

And the pre-compiled Go plugin Impl(this is only used for plugins that are written in Go):

// streaming/plugins/abci/{
    plugin_version
}/impl/plugin.go

// Plugins are pre-compiled and loaded by the plugin system

// ABCIListener is the implementation of the baseapp.ABCIListener interface
type ABCIListener struct{
}

func (m *ABCIListenerPlugin)

ListenFinalizeBlock(ctx context.Context, req abci.RequestFinalizeBlock, res abci.ResponseFinalizeBlock)

error {
    // send data to external system
}

func (m *ABCIListenerPlugin)

ListenCommit(ctx context.Context, res abci.ResponseCommit, changeSet []store.StoreKVPair)

error {
    // send data to external system
}

func main() {
    plugin.Serve(&plugin.ServeConfig{
    HandshakeConfig: grpc_abci_v1.Handshake,
    Plugins: map[string]plugin.Plugin{
           "grpc_plugin_v1": &grpc_abci_v1.ABCIListenerGRPCPlugin{
    Impl: &ABCIListenerPlugin{
}},
},

        // A non-nil value here enables gRPC serving for this streaming...
        GRPCServer: plugin.DefaultGRPCServer,
})
}

We will introduce a plugin loading system that will return (interface{}, error). This provides the advantage of using versioned plugins where the plugin interface and gRPC protocol change over time. In addition, it allows for building independent plugin that can expose different parts of the system over gRPC.

func NewStreamingPlugin(name string, logLevel string) (interface{
}, error) {
    logger := hclog.New(&hclog.LoggerOptions{
    Output: hclog.DefaultOutput,
    Level:  toHclogLevel(logLevel),
    Name:   fmt.Sprintf("plugin.%s", name),
})

    // We're a host. Start by launching the streaming process.
    env := os.Getenv(GetPluginEnvKey(name))
    client := plugin.NewClient(&plugin.ClientConfig{
    HandshakeConfig: HandshakeMap[name],
    Plugins:         PluginMap,
    Cmd:             exec.Command("sh", "-c", env),
    Logger:          logger,
    AllowedProtocols: []plugin.Protocol{
    plugin.ProtocolNetRPC, plugin.ProtocolGRPC
},
})

    // Connect via RPC
    rpcClient, err := client.Client()
    if err != nil {
    return nil, err
}

    // Request streaming plugin
    return rpcClient.Dispense(name)
}

We propose a RegisterStreamingPlugin function for the App to register NewStreamingPlugins with the App’s BaseApp. Streaming plugins can be of Any type; therefore, the function takes in an interface vs a concrete type. For example, we could have plugins of ABCIListener, WasmListener or IBCListener. Note that RegisterStreamingPluing function is helper function and not a requirement. Plugin registration can easily be moved from the App to the BaseApp directly.

// baseapp/streaming.go

// RegisterStreamingPlugin registers streaming plugins with the App.
// This method returns an error if a plugin is not supported.
func RegisterStreamingPlugin(
    bApp *BaseApp,
    appOpts servertypes.AppOptions,
    keys map[string]*types.KVStoreKey,
    streamingPlugin interface{
},
)

error {
    switch t := streamingPlugin.(type) {
    case ABCIListener:
        registerABCIListenerPlugin(bApp, appOpts, keys, t)

default:
        return fmt.Errorf("unexpected plugin type %T", t)
}

return nil
}

func registerABCIListenerPlugin(
    bApp *BaseApp,
    appOpts servertypes.AppOptions,
    keys map[string]*store.KVStoreKey,
    abciListener ABCIListener,
) {
    asyncKey := fmt.Sprintf("%s.%s.%s", StreamingTomlKey, StreamingABCITomlKey, StreamingABCIAsync)
    async := cast.ToBool(appOpts.Get(asyncKey))
    stopNodeOnErrKey := fmt.Sprintf("%s.%s.%s", StreamingTomlKey, StreamingABCITomlKey, StreamingABCIStopNodeOnErrTomlKey)
    stopNodeOnErr := cast.ToBool(appOpts.Get(stopNodeOnErrKey))
    keysKey := fmt.Sprintf("%s.%s.%s", StreamingTomlKey, StreamingABCITomlKey, StreamingABCIKeysTomlKey)
    exposeKeysStr := cast.ToStringSlice(appOpts.Get(keysKey))
    exposedKeys := exposeStoreKeysSorted(exposeKeysStr, keys)

bApp.cms.AddListeners(exposedKeys)

app.SetStreamingManager(
		storetypes.StreamingManager{
    ABCIListeners: []storetypes.ABCIListener{
    abciListener
},
    StopNodeOnErr: stopNodeOnErr,
},
	)
}

func exposeAll(list []string)

bool {
    for _, ele := range list {
    if ele == "*" {
    return true
}
 
}

return false
}

func exposeStoreKeys(keysStr []string, keys map[string]*types.KVStoreKey) []types.StoreKey {
    var exposeStoreKeys []types.StoreKey
    if exposeAll(keysStr) {
    exposeStoreKeys = make([]types.StoreKey, 0, len(keys))
    for _, storeKey := range keys {
    exposeStoreKeys = append(exposeStoreKeys, storeKey)
}
 
}

else {
    exposeStoreKeys = make([]types.StoreKey, 0, len(keysStr))
    for _, keyStr := range keysStr {
    if storeKey, ok := keys[keyStr]; ok {
    exposeStoreKeys = append(exposeStoreKeys, storeKey)
}
 
}
 
}
    // sort storeKeys for deterministic output
    sort.SliceStable(exposeStoreKeys, func(i, j int)

bool {
    return exposeStoreKeys[i].Name() < exposeStoreKeys[j].Name()
})

return exposeStoreKeys
}

The NewStreamingPlugin and RegisterStreamingPlugin functions are used to register a plugin with the App’s BaseApp. e.g. in NewSimApp:

func NewSimApp(
    logger log.Logger,
    db dbm.DB,
    traceStore io.Writer,
    loadLatest bool,
    appOpts servertypes.AppOptions,
    baseAppOptions ...func(*baseapp.BaseApp),
) *SimApp {

    ...
    keys := sdk.NewKVStoreKeys(
       authtypes.StoreKey, banktypes.StoreKey, stakingtypes.StoreKey,
       minttypes.StoreKey, distrtypes.StoreKey, slashingtypes.StoreKey,
       govtypes.StoreKey, paramstypes.StoreKey, ibchost.StoreKey, upgradetypes.StoreKey,
       evidencetypes.StoreKey, ibctransfertypes.StoreKey, capabilitytypes.StoreKey,
    )

    ...

    // register streaming services
    streamingCfg := cast.ToStringMap(appOpts.Get(baseapp.StreamingTomlKey))
    for service := range streamingCfg {
    pluginKey := fmt.Sprintf("%s.%s.%s", baseapp.StreamingTomlKey, service, baseapp.StreamingPluginTomlKey)
    pluginName := strings.TrimSpace(cast.ToString(appOpts.Get(pluginKey)))
    if len(pluginName) > 0 {
    logLevel := cast.ToString(appOpts.Get(flags.FlagLogLevel))

plugin, err := streaming.NewStreamingPlugin(pluginName, logLevel)
    if err != nil {
    tmos.Exit(err.Error())
}
    if err := baseapp.RegisterStreamingPlugin(bApp, appOpts, keys, plugin); err != nil {
    tmos.Exit(err.Error())
}
 
}
 
}

return app

Configuration

The plugin system will be configured within an App’s TOML configuration files.

# gRPC streaming
[streaming]

# ABCI streaming service
[streaming.abci]

# The plugin version to use for ABCI listening
plugin = "abci_v1"

# List of kv store keys to listen to for state changes.
# Set to ["*"] to expose all keys.
keys = ["*"]

# Enable abciListeners to run asynchronously.
# When abciListenersAsync=false and stopNodeOnABCIListenerErr=false listeners will run synchronized but will not stop the node.
# When abciListenersAsync=true stopNodeOnABCIListenerErr will be ignored.
async = false

# Whether to stop the node on message deliver error.
stop-node-on-err = true

There will be four parameters for configuring ABCIListener plugin: streaming.abci.plugin, streaming.abci.keys, streaming.abci.async and streaming.abci.stop-node-on-err. streaming.abci.plugin is the name of the plugin we want to use for streaming, streaming.abci.keys is a set of store keys for stores it listens to, streaming.abci.async is bool enabling asynchronous listening and streaming.abci.stop-node-on-err is a bool that stops the node when true and when operating on synchronized mode streaming.abci.async=false. Note that streaming.abci.stop-node-on-err=true will be ignored if streaming.abci.async=true. The configuration above support additional streaming plugins by adding the plugin to the [streaming] configuration section and registering the plugin with RegisterStreamingPlugin helper function. Note the that each plugin must include streaming.{service}.plugin property as it is a requirement for doing the lookup and registration of the plugin with the App. All other properties are unique to the individual services.

Encoding and decoding streams

ADR-038 introduces the interfaces and types for streaming state changes out from KVStores, associating this data with their related ABCI requests and responses, and registering a service for consuming this data and streaming it to some destination in a final format. Instead of prescribing a final data format in this ADR, it is left to a specific plugin implementation to define and document this format. We take this approach because flexibility in the final format is necessary to support a wide range of streaming service plugins. For example, the data format for a streaming service that writes the data out to a set of files will differ from the data format that is written to a Kafka topic.

Consequences

These changes will provide a means of subscribing to KVStore state changes in real time.

Backwards Compatibility

This ADR changes the CommitMultiStore interface, implementations supporting the previous version of this interface will not support the new one

Positive

Ability to listen to KVStore state changes in real time and expose these events to external consumers

Negative

Changes CommitMultiStore interface and its implementations

Neutral

Introduces additional- but optional- complexity to configuring and running a cosmos application
If an application developer opts to use these features to expose data, they need to be aware of the ramifications/risks of that data exposure as it pertains to the specifics of their application

​变更记录

​状态

​摘要

​背景

​决策

​监听

​ListenKVStore

​MultiStore 接口更新

​MultiStore 实现更新

​暴露数据

​流式服务

​BaseApp 注册

​ABCI 事件钩子

​Go 插件系统

​配置

​Whether to stop the node on message deliver error.

​ListenKVStore

​MultiStore interface updates

​MultiStore implementation updates

​Exposing the data

​Streaming Service

​BaseApp Registration

​ABCI Event Hooks

​Go Plugin System

​Configuration

​Encoding and decoding streams

​Consequences

​Backwards Compatibility

​Positive

​Negative

​Neutral

变更记录

状态

摘要

背景

决策

监听

ListenKVStore

MultiStore 接口更新

MultiStore 实现更新

暴露数据

流式服务

BaseApp 注册

ABCI 事件钩子

Go 插件系统

配置

Whether to stop the node on message deliver error.

ListenKVStore

MultiStore interface updates

MultiStore implementation updates

Exposing the data

Streaming Service

BaseApp Registration

ABCI Event Hooks

Go Plugin System

Configuration

Encoding and decoding streams

Consequences

Backwards Compatibility

Positive

Negative

Neutral