Repository: celestiaorg/celestia-node
Pull request #59 (2021-09-13 14:58:22)

block: create `Fetcher` interface
This PR introduces a block `Fetcher` interface that will encompass the behaviour necessary for the `BlockService` to get new blocks either directly from a Core node or from the Celestia network.

It also adds an implementation of `Fetcher` for the Core `Client`

TODO 
- [x] Add `BlockService` to `Full` node

Related to #25 and #26 

renaynay: re https://github.com/celestiaorg/celestia-node/pull/59#discussion_r710060177 @Wondertan -- because this channel is specifically to listen for `new` block events.

Patch:
diff --git a/core/fetcher.go b/core/fetcher.go
new file mode 100644
index 00000000..33f2a620
--- /dev/null
+++ b/core/fetcher.go
@@ -0,0 +1,94 @@
+package core
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/celestiaorg/celestia-core/types"
+	"github.com/celestiaorg/celestia-node/service/block"
+)
+
+const newBlockSubscriber = "NewBlock/Events"
+
+var newBlockEventQuery = types.QueryForEvent(types.EventNewBlock).String()
+
+type BlockFetcher struct {
+	client Client
+
+	newBlockCh chan *block.Raw
+}
+
+// NewBlockFetcher returns a new `BlockFetcher`.
+func NewBlockFetcher(client Client) *BlockFetcher {
+	return &BlockFetcher{
+		client: client,
+	}
+}
+
+// GetBlock queries Core for a `Block` at the given height.
+func (f *BlockFetcher) GetBlock(ctx context.Context, height *int64) (*block.Raw, error) {
+	raw, err := f.client.Block(ctx, height)
+	if err != nil {
+		return nil, err
+	}
+	return raw.Block, nil
+}
+
+// SubscribeNewBlockEvent subscribes to new block events from Core, returning
+// a new block event channel on success.
+func (f *BlockFetcher) SubscribeNewBlockEvent(ctx context.Context) (<-chan *block.Raw, error) {
+	// start the client if not started yet
+	if !f.client.IsRunning() {
+		return nil, fmt.Errorf("client not running")
+	}
+	eventChan, err := f.client.Subscribe(ctx, newBlockSubscriber, newBlockEventQuery)
+	if err != nil {
+		return nil, err
+	}
+
+	// create a wrapper channel for translating ResultEvent to "raw" block
+	if f.newBlockCh != nil {
+		return nil, fmt.Errorf("new block event channel exists")
+	}
+
+	f.newBlockCh = make(chan *block.Raw)
+
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case newEvent, ok := <-eventChan:
+				if !ok {
+					return
+				}
+				newBlock, ok := newEvent.Data.(types.EventDataNewBlock)
+				if !ok {
+					log.Warnf("unexpected event: %v", newEvent)
+					continue
+				}
+				select {
+				case f.newBlockCh <- newBlock.Block:
+				case <-ctx.Done():
+					return
+				}
+			}
+		}
+	}()
+
+	return f.newBlockCh, nil
+}
+
+// UnsubscribeNewBlockEvent stops the subscription to new block events from Core.
+func (f *BlockFetcher) UnsubscribeNewBlockEvent(ctx context.Context) error {
+	// close the new block channel
+	if f.newBlockCh == nil {
+		return fmt.Errorf("no new block event channel found")
+	}
+	defer func() {
+		close(f.newBlockCh)
+		f.newBlockCh = nil
+	}()
+
+	return f.client.Unsubscribe(ctx, newBlockSubscriber, newBlockEventQuery)
+}
diff --git a/node/full.go b/node/full.go
index 410e1fbd..57d35516 100644
--- a/node/full.go
+++ b/node/full.go
@@ -5,8 +5,10 @@ import (
 
 	"go.uber.org/fx"
 
+	coreclient "github.com/celestiaorg/celestia-node/core"
 	"github.com/celestiaorg/celestia-node/node/core"
 	"github.com/celestiaorg/celestia-node/node/p2p"
+	"github.com/celestiaorg/celestia-node/service/block"
 )
 
 // NewFull assembles a new Full Node from required components.
@@ -25,7 +27,11 @@ func fullComponents(cfg *Config) fx.Option {
 		fx.Provide(func() *Config {
 			return cfg
 		}),
-
+		// provide the block Fetcher
+		fx.Provide(func(client coreclient.Client) block.Fetcher {
+			return coreclient.NewBlockFetcher(client)
+		}),
+		fx.Provide(block.NewBlockService),
 		// components
 		p2p.Components(cfg.P2P),
 		core.Components(cfg.Core),
diff --git a/node/node.go b/node/node.go
index 54fa22f9..ffabc9e4 100644
--- a/node/node.go
+++ b/node/node.go
@@ -12,6 +12,7 @@ import (
 	"go.uber.org/fx"
 
 	"github.com/celestiaorg/celestia-node/core"
+	"github.com/celestiaorg/celestia-node/service/block"
 )
 
 var log = logging.Logger("node")
@@ -38,6 +39,12 @@ type Node struct {
 	DataExchange exchange.Interface
 	// p2p protocols
 	PubSub *pubsub.PubSub
+	// BlockService provides access to the node's Block Service
+	// TODO @renaynay: I don't like the concept of storing individual services on the node,
+	// TODO maybe create a struct in full.go that contains `FullServices` (all expected services to be running on a
+	// TODO full node) and in light, same thing `LightServices` (all expected services to be running in a light node.
+	// TODO `FullServices` can include `LightServices` + other services.
+	BlockServ *block.Service `optional:"true"`
 }
 
 // newNode creates a new Node from given DI options.
diff --git a/service/block/interface.go b/service/block/interface.go
new file mode 100644
index 00000000..18e02745
--- /dev/null
+++ b/service/block/interface.go
@@ -0,0 +1,12 @@
+package block
+
+import (
+	"context"
+)
+
+// Fetcher encompasses the behavior necessary to fetch new "raw" blocks.
+type Fetcher interface {
+	GetBlock(ctx context.Context, height *int64) (*Raw, error)
+	SubscribeNewBlockEvent(ctx context.Context) (<-chan *Raw, error)
+	UnsubscribeNewBlockEvent(ctx context.Context) error
+}
diff --git a/service/block/service.go b/service/block/service.go
new file mode 100644
index 00000000..eb48891a
--- /dev/null
+++ b/service/block/service.go
@@ -0,0 +1,34 @@
+package block
+
+import (
+	"context"
+)
+
+// Service represents the Block service that can be started / stopped on a `Full` node.
+// Service contains 4 main functionalities:
+//		1. Fetching "raw" blocks from either Celestia Core or other Celestia Full nodes.
+// 		2. Erasure coding the "raw" blocks and producing a DataAvailabilityHeader + verifying the Data root.
+// 		3. Storing erasure coded blocks.
+// 		4. Serving erasure coded blocks to other `Full` node peers.
+type Service struct {
+	fetcher Fetcher
+}
+
+// NewBlockService creates a new instance of block Service.
+func NewBlockService(fetcher Fetcher) *Service {
+	return &Service{
+		fetcher: fetcher,
+	}
+}
+
+// Start starts the block Service.
+// TODO @renaynay: make sure `Start` eventually has the same signature as `Stop`
+func (s *Service) Start(ctx context.Context) (<-chan *Raw, error) {
+	// TODO @renaynay: this will eventually be self contained within the block package
+	return s.fetcher.SubscribeNewBlockEvent(ctx)
+}
+
+// Stop stops the block Service.
+func (s *Service) Stop(ctx context.Context) error {
+	return s.fetcher.UnsubscribeNewBlockEvent(ctx)
+}
diff --git a/service/block/types.go b/service/block/types.go
new file mode 100644
index 00000000..b484cc45
--- /dev/null
+++ b/service/block/types.go
@@ -0,0 +1,9 @@
+package block
+
+import (
+	core "github.com/celestiaorg/celestia-core/types"
+)
+
+// Raw is an alias to a "raw" Core block. It is "raw" because
+// it is still awaiting erasure coding.
+type Raw = core.Block


Test patch:
diff --git a/core/client_test.go b/core/client_test.go
index de55ebcd..c837d7e0 100644
--- a/core/client_test.go
+++ b/core/client_test.go
@@ -4,14 +4,10 @@ import (
 	"context"
 	"testing"
 
-	"github.com/celestiaorg/celestia-core/types"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
-var newBlockSubscriber = "NewBlock/Events"
-var newBlockEventQuery = types.QueryForEvent(types.EventNewBlock).String()
-
 func TestEmbeddedClientLifecycle(t *testing.T) {
 	client := MockEmbeddedClient()
 	require.NoError(t, client.Stop())
diff --git a/core/fetcher_test.go b/core/fetcher_test.go
new file mode 100644
index 00000000..89cc10ac
--- /dev/null
+++ b/core/fetcher_test.go
@@ -0,0 +1,33 @@
+package core
+
+import (
+	"context"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBlockFetcher_GetBlock_and_SubscribeNewBlockEvent(t *testing.T) {
+	client := MockEmbeddedClient()
+	fetcher := NewBlockFetcher(client)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	t.Cleanup(cancel)
+
+	// generate some blocks
+	newBlockChan, err := fetcher.SubscribeNewBlockEvent(ctx)
+	require.NoError(t, err)
+
+	for i := 1; i < 3; i++ {
+		newBlockFromChan := <-newBlockChan
+
+		block, err := fetcher.GetBlock(ctx, nil)
+		require.NoError(t, err)
+
+		assert.Equal(t, newBlockFromChan, block)
+	}
+
+	require.NoError(t, fetcher.UnsubscribeNewBlockEvent(ctx))
+	require.NoError(t, client.Stop())
+}
diff --git a/node/full_test.go b/node/full_test.go
index 2fe8d013..9713b459 100644
--- a/node/full_test.go
+++ b/node/full_test.go
@@ -40,6 +40,7 @@ func TestFullLifecycle(t *testing.T) {
 	require.NotZero(t, node.Type)
 	require.NotNil(t, node.Host)
 	require.NotNil(t, node.CoreClient)
+	require.NotNil(t, node.BlockServ)
 
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)