erigon-pulse/direct/eth_backend_client.go

233 lines
7.3 KiB
Go

/*
Copyright 2021 Erigon contributors
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package direct
import (
"context"
"io"
"github.com/ledgerwatch/erigon-lib/gointerfaces/remote"
"github.com/ledgerwatch/erigon-lib/gointerfaces/types"
"google.golang.org/grpc"
"google.golang.org/protobuf/types/known/emptypb"
)
type EthBackendClientDirect struct {
server remote.ETHBACKENDServer
}
func NewEthBackendClientDirect(server remote.ETHBACKENDServer) *EthBackendClientDirect {
return &EthBackendClientDirect{server: server}
}
func (s *EthBackendClientDirect) Etherbase(ctx context.Context, in *remote.EtherbaseRequest, opts ...grpc.CallOption) (*remote.EtherbaseReply, error) {
return s.server.Etherbase(ctx, in)
}
func (s *EthBackendClientDirect) NetVersion(ctx context.Context, in *remote.NetVersionRequest, opts ...grpc.CallOption) (*remote.NetVersionReply, error) {
return s.server.NetVersion(ctx, in)
}
func (s *EthBackendClientDirect) NetPeerCount(ctx context.Context, in *remote.NetPeerCountRequest, opts ...grpc.CallOption) (*remote.NetPeerCountReply, error) {
return s.server.NetPeerCount(ctx, in)
}
func (s *EthBackendClientDirect) EngineNewPayload(ctx context.Context, in *types.ExecutionPayload, opts ...grpc.CallOption) (*remote.EnginePayloadStatus, error) {
return s.server.EngineNewPayload(ctx, in)
}
func (s *EthBackendClientDirect) EngineForkChoiceUpdated(ctx context.Context, in *remote.EngineForkChoiceUpdatedRequest, opts ...grpc.CallOption) (*remote.EngineForkChoiceUpdatedResponse, error) {
return s.server.EngineForkChoiceUpdated(ctx, in)
}
func (s *EthBackendClientDirect) EngineGetPayload(ctx context.Context, in *remote.EngineGetPayloadRequest, opts ...grpc.CallOption) (*remote.EngineGetPayloadResponse, error) {
return s.server.EngineGetPayload(ctx, in)
}
func (s *EthBackendClientDirect) EngineGetPayloadBodiesByHashV1(ctx context.Context, in *remote.EngineGetPayloadBodiesByHashV1Request, opts ...grpc.CallOption) (*remote.EngineGetPayloadBodiesV1Response, error) {
return s.server.EngineGetPayloadBodiesByHashV1(ctx, in)
}
func (s *EthBackendClientDirect) EngineGetPayloadBodiesByRangeV1(ctx context.Context, in *remote.EngineGetPayloadBodiesByRangeV1Request, opts ...grpc.CallOption) (*remote.EngineGetPayloadBodiesV1Response, error) {
return s.server.EngineGetPayloadBodiesByRangeV1(ctx, in)
}
func (s *EthBackendClientDirect) Version(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*types.VersionReply, error) {
return s.server.Version(ctx, in)
}
func (s *EthBackendClientDirect) ProtocolVersion(ctx context.Context, in *remote.ProtocolVersionRequest, opts ...grpc.CallOption) (*remote.ProtocolVersionReply, error) {
return s.server.ProtocolVersion(ctx, in)
}
func (s *EthBackendClientDirect) ClientVersion(ctx context.Context, in *remote.ClientVersionRequest, opts ...grpc.CallOption) (*remote.ClientVersionReply, error) {
return s.server.ClientVersion(ctx, in)
}
// -- start Subscribe
func (s *EthBackendClientDirect) Subscribe(ctx context.Context, in *remote.SubscribeRequest, opts ...grpc.CallOption) (remote.ETHBACKEND_SubscribeClient, error) {
ch := make(chan *subscribeReply, 16384)
streamServer := &SubscribeStreamS{ch: ch, ctx: ctx}
go func() {
defer close(ch)
streamServer.Err(s.server.Subscribe(in, streamServer))
}()
return &SubscribeStreamC{ch: ch, ctx: ctx}, nil
}
type subscribeReply struct {
r *remote.SubscribeReply
err error
}
type SubscribeStreamS struct {
ch chan *subscribeReply
ctx context.Context
grpc.ServerStream
}
func (s *SubscribeStreamS) Send(m *remote.SubscribeReply) error {
s.ch <- &subscribeReply{r: m}
return nil
}
func (s *SubscribeStreamS) Context() context.Context { return s.ctx }
func (s *SubscribeStreamS) Err(err error) {
if err == nil {
return
}
s.ch <- &subscribeReply{err: err}
}
type SubscribeStreamC struct {
ch chan *subscribeReply
ctx context.Context
grpc.ClientStream
}
func (c *SubscribeStreamC) Recv() (*remote.SubscribeReply, error) {
m, ok := <-c.ch
if !ok || m == nil {
return nil, io.EOF
}
return m.r, m.err
}
func (c *SubscribeStreamC) Context() context.Context { return c.ctx }
// -- end Subscribe
// -- SubscribeLogs
func (s *EthBackendClientDirect) SubscribeLogs(ctx context.Context, opts ...grpc.CallOption) (remote.ETHBACKEND_SubscribeLogsClient, error) {
subscribeLogsRequestChan := make(chan *subscribeLogsRequest, 16384)
subscribeLogsReplyChan := make(chan *subscribeLogsReply, 16384)
srv := &SubscribeLogsStreamS{
chSend: subscribeLogsReplyChan,
chRecv: subscribeLogsRequestChan,
ctx: ctx,
}
go func() {
defer close(subscribeLogsRequestChan)
defer close(subscribeLogsReplyChan)
srv.Err(s.server.SubscribeLogs(srv))
}()
cli := &SubscribeLogsStreamC{
chSend: subscribeLogsRequestChan,
chRecv: subscribeLogsReplyChan,
ctx: ctx,
}
return cli, nil
}
type SubscribeLogsStreamS struct {
chSend chan *subscribeLogsReply
chRecv chan *subscribeLogsRequest
ctx context.Context
grpc.ServerStream
}
type subscribeLogsReply struct {
r *remote.SubscribeLogsReply
err error
}
type subscribeLogsRequest struct {
r *remote.LogsFilterRequest
err error
}
func (s *SubscribeLogsStreamS) Send(m *remote.SubscribeLogsReply) error {
s.chSend <- &subscribeLogsReply{r: m}
return nil
}
func (s *SubscribeLogsStreamS) Recv() (*remote.LogsFilterRequest, error) {
m, ok := <-s.chRecv
if !ok || m == nil {
return nil, io.EOF
}
return m.r, m.err
}
func (s *SubscribeLogsStreamS) Err(err error) {
if err == nil {
return
}
s.chSend <- &subscribeLogsReply{err: err}
}
type SubscribeLogsStreamC struct {
chSend chan *subscribeLogsRequest
chRecv chan *subscribeLogsReply
ctx context.Context
grpc.ClientStream
}
func (c *SubscribeLogsStreamC) Send(m *remote.LogsFilterRequest) error {
c.chSend <- &subscribeLogsRequest{r: m}
return nil
}
func (c *SubscribeLogsStreamC) Recv() (*remote.SubscribeLogsReply, error) {
m, ok := <-c.chRecv
if !ok || m == nil {
return nil, io.EOF
}
return m.r, m.err
}
// -- end SubscribeLogs
func (s *EthBackendClientDirect) Block(ctx context.Context, in *remote.BlockRequest, opts ...grpc.CallOption) (*remote.BlockReply, error) {
return s.server.Block(ctx, in)
}
func (s *EthBackendClientDirect) TxnLookup(ctx context.Context, in *remote.TxnLookupRequest, opts ...grpc.CallOption) (*remote.TxnLookupReply, error) {
return s.server.TxnLookup(ctx, in)
}
func (s *EthBackendClientDirect) NodeInfo(ctx context.Context, in *remote.NodesInfoRequest, opts ...grpc.CallOption) (*remote.NodesInfoReply, error) {
return s.server.NodeInfo(ctx, in)
}
func (s *EthBackendClientDirect) Peers(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*remote.PeersReply, error) {
return s.server.Peers(ctx, in)
}
func (s *EthBackendClientDirect) PendingBlock(ctx context.Context, in *emptypb.Empty, opts ...grpc.CallOption) (*remote.PendingBlockReply, error) {
return s.server.PendingBlock(ctx, in)
}