deep-stack/ipld-eth-server
4
0
Fork 0
forked from cerc-io/ipld-eth-server
Code Pull Requests Activity

publisher unit test

Browse Source
This commit is contained in:
Ian Norden
2019-12-02 13:24:58 -06:00
parent d79cc90cb2
commit 35c8f3561a
16 changed files with 225 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/seed_node/api.go
+6 -6
View File
@@ -34,13 +34,13 @@ const APIVersion = "0.0.1"
// PublicSeedNodeAPI is the public api for the seed node
type PublicSeedNodeAPI struct {
snp Processor
sni NodeInterface
}
// NewPublicSeedNodeAPI creates a new PublicSeedNodeAPI with the provided underlying SyncPublishScreenAndServe process
func NewPublicSeedNodeAPI(seedNodeProcessor Processor) *PublicSeedNodeAPI {
func NewPublicSeedNodeAPI(seedNodeInterface NodeInterface) *PublicSeedNodeAPI {
return &PublicSeedNodeAPI{
snp: seedNodeProcessor,
sni: seedNodeInterface,
}
}
@@ -59,7 +59,7 @@ func (api *PublicSeedNodeAPI) Stream(ctx context.Context, streamFilters config.S
// subscribe to events from the SyncPublishScreenAndServe service
payloadChannel := make(chan streamer.SeedNodePayload, payloadChanBufferSize)
quitChan := make(chan bool, 1)
go api.snp.Subscribe(rpcSub.ID, payloadChannel, quitChan, streamFilters)
go api.sni.Subscribe(rpcSub.ID, payloadChannel, quitChan, streamFilters)
// loop and await state diff payloads and relay them to the subscriber with then notifier
for {
@@ -67,11 +67,11 @@ func (api *PublicSeedNodeAPI) Stream(ctx context.Context, streamFilters config.S
case packet := <-payloadChannel:
if notifyErr := notifier.Notify(rpcSub.ID, packet); notifyErr != nil {
log.Error("Failed to send state diff packet", "err", notifyErr)
api.snp.Unsubscribe(rpcSub.ID)
api.sni.Unsubscribe(rpcSub.ID)
return
}
case <-rpcSub.Err():
api.snp.Unsubscribe(rpcSub.ID)
api.sni.Unsubscribe(rpcSub.ID)
return
case <-quitChan:
// don't need to unsubscribe, SyncPublishScreenAndServe service does so before sending the quit signal
pkg/seed_node/filterer.go
+1 -1
View File
@@ -24,8 +24,8 @@ import (
"github.com/ethereum/go-ethereum/rlp"
"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
)
// ResponseFilterer is the inteface used to screen eth data and package appropriate data into a response payload
pkg/seed_node/repository.go
+1 -1
View File
@@ -20,8 +20,8 @@ import (
"github.com/jmoiron/sqlx"
"github.com/lib/pq"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
)
// CIDRepository is an interface for indexing ipfs.CIDPayloads
pkg/seed_node/retreiver.go
+1 -1
View File
@@ -23,9 +23,9 @@ import (
"github.com/lib/pq"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
)
// CIDRetriever is the interface for retrieving CIDs from the Postgres cache
pkg/seed_node/seed_node_suite_test.go
+1 -1
View File
@@ -20,9 +20,9 @@ import (
"io/ioutil"
"testing"
"github.com/sirupsen/logrus"
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/sirupsen/logrus"
)
func TestSeedNode(t *testing.T) {
pkg/seed_node/service.go
+12 -10
View File
@@ -19,30 +19,29 @@ package seed_node
import (
"sync"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/node"
"github.com/ethereum/go-ethereum/p2p"
"github.com/ethereum/go-ethereum/params"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/rpc"
"github.com/ethereum/go-ethereum/statediff"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/libraries/shared/streamer"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/core"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
)
const payloadChanBufferSize = 20000 // the max eth sub buffer size
const workerPoolSize = 1
// Processor is the top level interface for streaming, converting to IPLDs, publishing,
// NodeInterface is the top level interface for streaming, converting to IPLDs, publishing,
// and indexing all Ethereum data; screening this data; and serving it up to subscribed clients
// This service is compatible with the Ethereum service interface (node.Service)
type Processor interface {
type NodeInterface interface {
// APIs(), Protocols(), Start() and Stop()
node.Service
// Main event loop for syncAndPublish processes
@@ -83,10 +82,12 @@ type Service struct {
Subscriptions map[common.Hash]map[rpc.ID]Subscription
// A mapping of subscription hash type to the corresponding StreamFilters
SubscriptionTypes map[common.Hash]config.Subscription
// Number of workers
WorkerPoolSize int
}
// NewProcessor creates a new Processor interface using an underlying Service struct
func NewProcessor(ipfsPath string, db *postgres.DB, ethClient core.EthClient, rpcClient core.RpcClient, qc chan bool) (Processor, error) {
// NewSeedNode creates a new seed_node.Interface using an underlying seed_node.Service struct
func NewSeedNode(ipfsPath string, db *postgres.DB, rpcClient core.RpcClient, qc chan bool, workers int) (NodeInterface, error) {
publisher, err := ipfs.NewIPLDPublisher(ipfsPath)
if err != nil {
return nil, err
@@ -108,6 +109,7 @@ func NewProcessor(ipfsPath string, db *postgres.DB, ethClient core.EthClient, rp
QuitChan: qc,
Subscriptions: make(map[common.Hash]map[rpc.ID]Subscription),
SubscriptionTypes: make(map[common.Hash]config.Subscription),
WorkerPoolSize: workers,
}, nil
}
@@ -140,10 +142,10 @@ func (sap *Service) SyncAndPublish(wg *sync.WaitGroup, screenAndServePayload cha
// Channels for forwarding data to the publishAndIndex workers
publishAndIndexPayload := make(chan ipfs.IPLDPayload, payloadChanBufferSize)
publishAndIndexQuit := make(chan bool, workerPoolSize)
publishAndIndexQuit := make(chan bool, sap.WorkerPoolSize)
// publishAndIndex worker pool to handle publishing and indexing concurrently, while
// limiting the number of Postgres connections we can possibly open so as to prevent error
for i := 0; i < workerPoolSize; i++ {
for i := 0; i < sap.WorkerPoolSize; i++ {
sap.publishAndIndex(i, publishAndIndexPayload, publishAndIndexQuit)
}
@@ -175,7 +177,7 @@ func (sap *Service) SyncAndPublish(wg *sync.WaitGroup, screenAndServePayload cha
default:
}
// Also forward a quit signal for each of the workers
for i := 0; i < workerPoolSize; i++ {
for i := 0; i < sap.WorkerPoolSize; i++ {
select {
case publishAndIndexQuit <- true:
default:
pkg/seed_node/service_test.go
+15 -14
View File
@@ -25,15 +25,15 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/seed_node"
mocks2 "github.com/vulcanize/vulcanizedb/libraries/shared/mocks"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
"github.com/vulcanize/vulcanizedb/pkg/seed_node"
mocks3 "github.com/vulcanize/vulcanizedb/pkg/seed_node/mocks"
)
var _ = Describe("Service", func() {
Describe("Loop", func() {
Describe("SyncAndPublish", func() {
It("Streams statediff.Payloads, converts them to IPLDPayloads, publishes IPLDPayloads, and indexes CIDPayloads", func() {
wg := new(sync.WaitGroup)
payloadChan := make(chan statediff.Payload, 1)
@@ -42,36 +42,37 @@ var _ = Describe("Service", func() {
ReturnErr: nil,
}
mockPublisher := &mocks.IPLDPublisher{
ReturnCIDPayload: &mocks.MockCIDPayload,
ReturnCIDPayload: mocks.MockCIDPayload,
ReturnErr: nil,
}
mockStreamer := &mocks2.StateDiffStreamer{
ReturnSub: &rpc.ClientSubscription{},
StreamPayloads: []statediff.Payload{
mocks.MockStatediffPayload,
mocks.MockStateDiffPayload,
},
ReturnErr: nil,
}
mockConverter := &mocks.PayloadConverter{
ReturnIPLDPayload: &mocks.MockIPLDPayload,
ReturnIPLDPayload: mocks.MockIPLDPayload,
ReturnErr: nil,
}
processor := &seed_node.Service{
Repository: mockCidRepo,
Publisher: mockPublisher,
Streamer: mockStreamer,
Converter: mockConverter,
PayloadChan: payloadChan,
QuitChan: quitChan,
Repository: mockCidRepo,
Publisher: mockPublisher,
Streamer: mockStreamer,
Converter: mockConverter,
PayloadChan: payloadChan,
QuitChan: quitChan,
WorkerPoolSize: 1,
}
err := processor.SyncAndPublish(wg, nil, nil)
Expect(err).ToNot(HaveOccurred())
time.Sleep(2 * time.Second)
quitChan <- true
wg.Wait()
Expect(mockConverter.PassedStatediffPayload).To(Equal(mocks.MockStatediffPayload))
Expect(mockCidRepo.PassedCIDPayload).To(Equal(&mocks.MockCIDPayload))
Expect(mockPublisher.PassedIPLDPayload).To(Equal(&mocks.MockIPLDPayload))
Expect(mockConverter.PassedStatediffPayload).To(Equal(mocks.MockStateDiffPayload))
Expect(mockCidRepo.PassedCIDPayload).To(Equal(mocks.MockCIDPayload))
Expect(mockPublisher.PassedIPLDPayload).To(Equal(mocks.MockIPLDPayload))
Expect(mockStreamer.PassedPayloadChan).To(Equal(payloadChan))
})
})