infra

Documentation

Index

• Variables
• func CreateTables(ctx context.Context, db DBPool) error
• func DropTables(ctx context.Context, db DBPool) error
• func NewAlloraClient(rpcURL string, timeout time.Duration) (domain.AlloraClientInterface, error)
• func NewFranzClient(seeds []string, user, password string) (*kgo.Client, error)
• func NewKafkaClient(client KafkaClient, router domain.TopicRouter) (domain.StreamingClient, error)
• func NewPgProcessedBlock(db DBPool) (domain.ProcessedBlockRepositoryInterface, error)
• func NewPgQueryResult(db DBPool) (domain.QueryResultRepositoryInterface, error)
• type AlloraClient
  • func (a *AlloraClient) GetBlockByHeight(ctx context.Context, height int64) (*coretypes.ResultBlock, error)
  • func (a *AlloraClient) GetBlockResults(ctx context.Context, height int64) (*coretypes.ResultBlockResults, error)
  • func (a *AlloraClient) GetHeader(ctx context.Context, height int64) (*coretypes.ResultHeader, error)
  • func (a *AlloraClient) GetLatestBlockHeight(ctx context.Context) (int64, error)
  • func (a *AlloraClient) Query(ctx context.Context, path string, data bytes.HexBytes) (*coretypes.ResultABCIQuery, error)
• type DBPool
• type KafkaClient
• type QueryResultRepositoryInfra
  • func (q *QueryResultRepositoryInfra) SaveQueryResult(ctx context.Context, key string, queryType string, metadata json.RawMessage, ...) error
• type RPCClient
• type RowInterface
• type TopicRouterImpl
  • func NewTopicRouter(mapping map[string]string) *TopicRouterImpl
  • func (tr *TopicRouterImpl) GetTopic(msgType string) (string, error)

Variables

var ErrUnknownMessageType = errors.New("unknown message type")

ErrUnknownMessageType is returned when the message type is not recognized.

Functions

func CreateTables

func CreateTables(ctx context.Context, db DBPool) error

func DropTables

func DropTables(ctx context.Context, db DBPool) error

func NewAlloraClient

func NewAlloraClient(rpcURL string, timeout time.Duration) (domain.AlloraClientInterface, error)

func NewFranzClient

func NewFranzClient(seeds []string, user, password string) (*kgo.Client, error)

func NewKafkaClient

func NewKafkaClient(client KafkaClient, router domain.TopicRouter) (domain.StreamingClient, error)

Update the constructor to accept a TopicRouter.

func NewPgProcessedBlock

func NewPgProcessedBlock(db DBPool) (domain.ProcessedBlockRepositoryInterface, error)

NewPgProcessedBlock creates a new instance of pgProcessedBlock

func NewPgQueryResult

func NewPgQueryResult(db DBPool) (domain.QueryResultRepositoryInterface, error)

Types

type AlloraClient

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

func (*AlloraClient) GetBlockByHeight

func (a *AlloraClient) GetBlockByHeight(ctx context.Context, height int64) (*coretypes.ResultBlock, error)

GetBlockByHeight implements domain.AlloraClientInterface.

func (*AlloraClient) GetBlockResults

func (a *AlloraClient) GetBlockResults(ctx context.Context, height int64) (*coretypes.ResultBlockResults, error)

GetBlockResults implements domain.AlloraClientInterface.

func (*AlloraClient) GetHeader

func (a *AlloraClient) GetHeader(ctx context.Context, height int64) (*coretypes.ResultHeader, error)

func (*AlloraClient) GetLatestBlockHeight

func (a *AlloraClient) GetLatestBlockHeight(ctx context.Context) (int64, error)

GetLatestBlockHeight implements domain.AlloraClientInterface.

func (*AlloraClient) Query

func (a *AlloraClient) Query(ctx context.Context, path string, data bytes.HexBytes) (*coretypes.ResultABCIQuery, error)

type DBPool

type DBPool interface {
	QueryRow(ctx context.Context, sql string, args ...interface{}) RowInterface
	Exec(ctx context.Context, sql string, arguments ...interface{}) (pgconn.CommandTag, error)
}

DBPool defines the interface for the database pool. This allows us to mock the database interactions in tests.

type KafkaClient

type KafkaClient interface {
	Produce(ctx context.Context, r *kgo.Record, promise func(*kgo.Record, error))
	Flush(ctx context.Context) error
	Close()
}

type QueryResultRepositoryInfra

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

func (*QueryResultRepositoryInfra) SaveQueryResult

func (q *QueryResultRepositoryInfra) SaveQueryResult(ctx context.Context, key string, queryType string, metadata json.RawMessage, value json.RawMessage) error

type RPCClient

type RPCClient interface {
	Block(ctx context.Context, height *int64) (*coretypes.ResultBlock, error)
	BlockResults(ctx context.Context, height *int64) (*coretypes.ResultBlockResults, error)
	Status(ctx context.Context) (*coretypes.ResultStatus, error)
	Header(ctx context.Context, height *int64) (*coretypes.ResultHeader, error)
	ABCIQuery(ctx context.Context, path string, data bytes.HexBytes) (*coretypes.ResultABCIQuery, error)
}

type RowInterface

type RowInterface = pgx.Row

type TopicRouterImpl

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

TopicRouterImpl determines the Kafka topic for a given message type.

func NewTopicRouter

func NewTopicRouter(mapping map[string]string) *TopicRouterImpl

NewTopicRouter initializes a new TopicRouter with the provided mappings.

func (*TopicRouterImpl) GetTopic

func (tr *TopicRouterImpl) GetTopic(msgType string) (string, error)

GetTopic returns the Kafka topic for the given message type.