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

fix backfill operations and dependency issue; hopefully travis will work now

Browse Source
This commit is contained in:
Ian Norden
2019-12-02 13:24:51 -06:00
parent 4baea2923c
commit 723c7c6244
93 changed files with 1832 additions and 8873 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/ipfs/api.go
+7 -2
View File
@@ -64,8 +64,13 @@ func (api *PublicSeedNodeAPI) Stream(ctx context.Context, streamFilters config.S
for {
select {
case packet := <-payloadChannel:
if err := notifier.Notify(rpcSub.ID, packet); err != nil {
log.Error("Failed to send state diff packet", "err", err)
if notifyErr := notifier.Notify(rpcSub.ID, packet); notifyErr != nil {
log.Error("Failed to send state diff packet", "err", notifyErr)
unSubErr := api.snp.Unsubscribe(rpcSub.ID)
if unSubErr != nil {
log.Error("Failed to unsubscribe from the state diff service", unSubErr)
}
return
}
case <-rpcSub.Err():
err := api.snp.Unsubscribe(rpcSub.ID)
pkg/ipfs/converter_test.go
+5 -6
View File
@@ -20,19 +20,18 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers/mocks"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
)
var _ = Describe("Converter", func() {
Describe("Convert", func() {
It("Converts StatediffPayloads into IPLDPayloads", func() {
mockConverter := mocks.PayloadConverter{}
mockConverter.ReturnIPLDPayload = &helpers.MockIPLDPayload
ipldPayload, err := mockConverter.Convert(helpers.MockStatediffPayload)
mockConverter.ReturnIPLDPayload = &mocks.MockIPLDPayload
ipldPayload, err := mockConverter.Convert(mocks.MockStatediffPayload)
Expect(err).ToNot(HaveOccurred())
Expect(ipldPayload).To(Equal(&helpers.MockIPLDPayload))
Expect(mockConverter.PassedStatediffPayload).To(Equal(helpers.MockStatediffPayload))
Expect(ipldPayload).To(Equal(&mocks.MockIPLDPayload))
Expect(mockConverter.PassedStatediffPayload).To(Equal(mocks.MockStatediffPayload))
})
})
})
pkg/ipfs/fetcher.go
+6 -1
View File
@@ -20,7 +20,6 @@ import (
"context"
"github.com/ethereum/go-ethereum/common"
"github.com/ipfs/go-block-format"
"github.com/ipfs/go-blockservice"
"github.com/ipfs/go-cid"
@@ -50,6 +49,7 @@ func NewIPLDFetcher(ipfsPath string) (*EthIPLDFetcher, error) {
// FetchCIDs is the exported method for fetching and returning all the cids passed in a CidWrapper
func (f *EthIPLDFetcher) FetchCIDs(cids CidWrapper) (*IpldWrapper, error) {
log.Debug("fetching iplds")
blocks := &IpldWrapper{
Headers: make([]blocks.Block, 0),
Transactions: make([]blocks.Block, 0),
@@ -85,6 +85,7 @@ func (f *EthIPLDFetcher) FetchCIDs(cids CidWrapper) (*IpldWrapper, error) {
// fetchHeaders fetches headers
// It uses the f.fetchBatch method
func (f *EthIPLDFetcher) fetchHeaders(cids CidWrapper, blocks *IpldWrapper) error {
log.Debug("fetching header iplds")
headerCids := make([]cid.Cid, 0, len(cids.Headers))
for _, c := range cids.Headers {
dc, err := cid.Decode(c)
@@ -103,6 +104,7 @@ func (f *EthIPLDFetcher) fetchHeaders(cids CidWrapper, blocks *IpldWrapper) erro
// fetchTrxs fetches transactions
// It uses the f.fetchBatch method
func (f *EthIPLDFetcher) fetchTrxs(cids CidWrapper, blocks *IpldWrapper) error {
log.Debug("fetching transaction iplds")
trxCids := make([]cid.Cid, 0, len(cids.Transactions))
for _, c := range cids.Transactions {
dc, err := cid.Decode(c)
@@ -121,6 +123,7 @@ func (f *EthIPLDFetcher) fetchTrxs(cids CidWrapper, blocks *IpldWrapper) error {
// fetchRcts fetches receipts
// It uses the f.fetchBatch method
func (f *EthIPLDFetcher) fetchRcts(cids CidWrapper, blocks *IpldWrapper) error {
log.Debug("fetching receipt iplds")
rctCids := make([]cid.Cid, 0, len(cids.Receipts))
for _, c := range cids.Receipts {
dc, err := cid.Decode(c)
@@ -140,6 +143,7 @@ func (f *EthIPLDFetcher) fetchRcts(cids CidWrapper, blocks *IpldWrapper) error {
// It uses the single f.fetch method instead of the batch fetch, because it
// needs to maintain the data's relation to state keys
func (f *EthIPLDFetcher) fetchState(cids CidWrapper, blocks *IpldWrapper) error {
log.Debug("fetching state iplds")
for _, stateNode := range cids.StateNodes {
if stateNode.CID == "" || stateNode.Key == "" {
continue
@@ -161,6 +165,7 @@ func (f *EthIPLDFetcher) fetchState(cids CidWrapper, blocks *IpldWrapper) error
// It uses the single f.fetch method instead of the batch fetch, because it
// needs to maintain the data's relation to state and storage keys
func (f *EthIPLDFetcher) fetchStorage(cids CidWrapper, blocks *IpldWrapper) error {
log.Debug("fetching storage iplds")
for _, storageNode := range cids.StorageNodes {
if storageNode.CID == "" || storageNode.Key == "" || storageNode.StateKey == "" {
continue
pkg/ipfs/helpers/mocks/api.go → pkg/ipfs/mocks/api.go
View File
pkg/ipfs/helpers/mocks/converter.go → pkg/ipfs/mocks/converter.go
View File
pkg/ipfs/helpers/mocks/publisher.go → pkg/ipfs/mocks/publisher.go
View File
pkg/ipfs/helpers/mocks/repository.go → pkg/ipfs/mocks/repository.go
View File
pkg/ipfs/helpers/mocks/screener.go → pkg/ipfs/mocks/screener.go
View File
pkg/ipfs/helpers/mocks/streamer.go → pkg/ipfs/mocks/streamer.go
View File
pkg/ipfs/helpers/test_data.go → pkg/ipfs/mocks/test_data.go
+3 -3
View File
@@ -14,20 +14,20 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package helpers
package mocks
import (
"errors"
"math/big"
"math/rand"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/rlp"
"github.com/ethereum/go-ethereum/statediff"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
)
// Test variables
pkg/ipfs/publisher_test.go
+5 -6
View File
@@ -20,19 +20,18 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers/mocks"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
)
var _ = Describe("Publisher", func() {
Describe("Publish", func() {
It("Publishes IPLDPayload to IPFS", func() {
mockPublisher := mocks.IPLDPublisher{}
mockPublisher.ReturnCIDPayload = &helpers.MockCIDPayload
cidPayload, err := mockPublisher.Publish(&helpers.MockIPLDPayload)
mockPublisher.ReturnCIDPayload = &mocks.MockCIDPayload
cidPayload, err := mockPublisher.Publish(&mocks.MockIPLDPayload)
Expect(err).ToNot(HaveOccurred())
Expect(cidPayload).To(Equal(&helpers.MockCIDPayload))
Expect(mockPublisher.PassedIPLDPayload).To(Equal(&helpers.MockIPLDPayload))
Expect(cidPayload).To(Equal(&mocks.MockCIDPayload))
Expect(mockPublisher.PassedIPLDPayload).To(Equal(&mocks.MockIPLDPayload))
})
})
})
pkg/ipfs/repository_test.go
+3 -4
View File
@@ -20,17 +20,16 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers/mocks"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
)
var _ = Describe("Repository", func() {
Describe("Index", func() {
It("Indexes CIDs against their metadata", func() {
mockRepo := mocks.CIDRepository{}
err := mockRepo.Index(&helpers.MockCIDPayload)
err := mockRepo.Index(&mocks.MockCIDPayload)
Expect(err).ToNot(HaveOccurred())
Expect(mockRepo.PassedCIDPayload).To(Equal(&helpers.MockCIDPayload))
Expect(mockRepo.PassedCIDPayload).To(Equal(&mocks.MockCIDPayload))
})
})
})
pkg/ipfs/resolver.go
-1
View File
@@ -18,7 +18,6 @@ package ipfs
import (
"github.com/ethereum/go-ethereum/common"
"github.com/ipfs/go-block-format"
)
pkg/ipfs/retreiver.go
+53 -38
View File
@@ -19,6 +19,8 @@ package ipfs
import (
"github.com/jmoiron/sqlx"
"github.com/lib/pq"
log "github.com/sirupsen/logrus"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/vulcanize/vulcanizedb/pkg/datastore/postgres"
)
@@ -49,6 +51,7 @@ func (ecr *EthCIDRetriever) GetLastBlockNumber() (int64, error) {
// RetrieveCIDs is used to retrieve all of the CIDs which conform to the passed StreamFilters
func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription) ([]CidWrapper, error) {
log.Debug("retrieving cids")
var endingBlock int64
var err error
if streamFilters.EndingBlock <= 0 || streamFilters.EndingBlock <= streamFilters.StartingBlock {
@@ -63,70 +66,68 @@ func (ecr *EthCIDRetriever) RetrieveCIDs(streamFilters config.Subscription) ([]C
return nil, err
}
for i := streamFilters.StartingBlock; i <= endingBlock; i++ {
cw := &CidWrapper{
BlockNumber: i,
Headers: make([]string, 0),
Transactions: make([]string, 0),
Receipts: make([]string, 0),
StateNodes: make([]StateNodeCID, 0),
StorageNodes: make([]StorageNodeCID, 0),
}
cw := CidWrapper{}
if !streamFilters.HeaderFilter.Off {
err = ecr.retrieveHeaderCIDs(tx, streamFilters, cw, i)
cw.Headers, err = ecr.retrieveHeaderCIDs(tx, streamFilters, i)
if err != nil {
tx.Rollback()
log.Error("header cid retrieval error")
return nil, err
}
}
var trxIds []int64
if !streamFilters.TrxFilter.Off {
trxIds, err = ecr.retrieveTrxCIDs(tx, streamFilters, cw, i)
cw.Transactions, trxIds, err = ecr.retrieveTrxCIDs(tx, streamFilters, i)
if err != nil {
tx.Rollback()
log.Error("transaction cid retrieval error")
return nil, err
}
}
if !streamFilters.ReceiptFilter.Off {
err = ecr.retrieveRctCIDs(tx, streamFilters, cw, i, trxIds)
cw.Receipts, err = ecr.retrieveRctCIDs(tx, streamFilters, i, trxIds)
if err != nil {
tx.Rollback()
log.Error("receipt cid retrieval error")
return nil, err
}
}
if !streamFilters.StateFilter.Off {
err = ecr.retrieveStateCIDs(tx, streamFilters, cw, i)
cw.StateNodes, err = ecr.retrieveStateCIDs(tx, streamFilters, i)
if err != nil {
tx.Rollback()
log.Error("state cid retrieval error")
return nil, err
}
}
if !streamFilters.StorageFilter.Off {
err = ecr.retrieveStorageCIDs(tx, streamFilters, cw, i)
cw.StorageNodes, err = ecr.retrieveStorageCIDs(tx, streamFilters, i)
if err != nil {
tx.Rollback()
log.Error("storage cid retrieval error")
return nil, err
}
}
cids = append(cids, *cw)
cids = append(cids, cw)
}
return cids, err
return cids, tx.Commit()
}
func (ecr *EthCIDRetriever) retrieveHeaderCIDs(tx *sqlx.Tx, streamFilters config.Subscription, cids *CidWrapper, blockNumber int64) error {
var pgStr string
if streamFilters.HeaderFilter.FinalOnly {
pgStr = `SELECT cid FROM header_cids
WHERE block_number = $1
AND final IS TRUE`
} else {
pgStr = `SELECT cid FROM header_cids
func (ecr *EthCIDRetriever) retrieveHeaderCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]string, error) {
log.Debug("retrieving header cids")
headers := make([]string, 0)
pgStr := `SELECT cid FROM header_cids
WHERE block_number = $1`
if streamFilters.HeaderFilter.FinalOnly {
pgStr += ` AND final IS TRUE`
}
return tx.Select(cids.Headers, pgStr, blockNumber)
err := tx.Select(&headers, pgStr, blockNumber)
return headers, err
}
func (ecr *EthCIDRetriever) retrieveTrxCIDs(tx *sqlx.Tx, streamFilters config.Subscription, cids *CidWrapper, blockNumber int64) ([]int64, error) {
func (ecr *EthCIDRetriever) retrieveTrxCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]string, []int64, error) {
log.Debug("retrieving transaction cids")
args := make([]interface{}, 0, 3)
type result struct {
ID int64 `db:"id"`
@@ -144,19 +145,21 @@ func (ecr *EthCIDRetriever) retrieveTrxCIDs(tx *sqlx.Tx, streamFilters config.Su
pgStr += ` AND transaction_cids.src = ANY($3::VARCHAR(66)[])`
args = append(args, pq.Array(streamFilters.TrxFilter.Src))
}
err := tx.Select(results, pgStr, args...)
err := tx.Select(&results, pgStr, args...)
if err != nil {
return nil, err
return nil, nil, err
}
ids := make([]int64, 0)
ids := make([]int64, 0, len(results))
cids := make([]string, 0, len(results))
for _, res := range results {
cids.Transactions = append(cids.Transactions, res.Cid)
cids = append(cids, res.Cid)
ids = append(ids, res.ID)
}
return ids, nil
return cids, ids, nil
}
func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Subscription, cids *CidWrapper, blockNumber int64, trxIds []int64) error {
func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64, trxIds []int64) ([]string, error) {
log.Debug("retrieving receipt cids")
args := make([]interface{}, 0, 2)
pgStr := `SELECT receipt_cids.cid FROM receipt_cids, transaction_cids, header_cids
WHERE receipt_cids.tx_id = transaction_cids.id
@@ -173,10 +176,13 @@ func (ecr *EthCIDRetriever) retrieveRctCIDs(tx *sqlx.Tx, streamFilters config.Su
} else {
pgStr += `)`
}
return tx.Select(cids.Receipts, pgStr, args...)
receiptCids := make([]string, 0)
err := tx.Select(&receiptCids, pgStr, args...)
return receiptCids, err
}
func (ecr *EthCIDRetriever) retrieveStateCIDs(tx *sqlx.Tx, streamFilters config.Subscription, cids *CidWrapper, blockNumber int64) error {
func (ecr *EthCIDRetriever) retrieveStateCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]StateNodeCID, error) {
log.Debug("retrieving state cids")
args := make([]interface{}, 0, 2)
pgStr := `SELECT state_cids.cid, state_cids.state_key FROM state_cids INNER JOIN header_cids ON (state_cids.header_id = header_cids.id)
WHERE header_cids.block_number = $1`
@@ -190,10 +196,16 @@ func (ecr *EthCIDRetriever) retrieveStateCIDs(tx *sqlx.Tx, streamFilters config.
pgStr += ` AND state_cids.state_key = ANY($2::VARCHAR(66)[])`
args = append(args, pq.Array(keys))
}
return tx.Select(cids.StateNodes, pgStr, args...)
if !streamFilters.StorageFilter.IntermediateNodes {
pgStr += ` AND state_cids.leaf = TRUE`
}
stateNodeCIDs := make([]StateNodeCID, 0)
err := tx.Select(&stateNodeCIDs, pgStr, args...)
return stateNodeCIDs, err
}
func (ecr *EthCIDRetriever) retrieveStorageCIDs(tx *sqlx.Tx, streamFilters config.Subscription, cids *CidWrapper, blockNumber int64) error {
func (ecr *EthCIDRetriever) retrieveStorageCIDs(tx *sqlx.Tx, streamFilters config.Subscription, blockNumber int64) ([]StorageNodeCID, error) {
log.Debug("retrieving storage cids")
args := make([]interface{}, 0, 3)
pgStr := `SELECT storage_cids.cid, state_cids.state_key, storage_cids.storage_key FROM storage_cids, state_cids, header_cids
WHERE storage_cids.state_id = state_cids.id
@@ -213,7 +225,10 @@ func (ecr *EthCIDRetriever) retrieveStorageCIDs(tx *sqlx.Tx, streamFilters confi
pgStr += ` AND storage_cids.storage_key = ANY($3::VARCHAR(66)[])`
args = append(args, pq.Array(streamFilters.StorageFilter.StorageKeys))
}
return tx.Select(cids.StorageNodes, pgStr, args...)
if !streamFilters.StorageFilter.IntermediateNodes {
pgStr += ` AND storage_cids.leaf = TRUE`
}
storageNodeCIDs := make([]StorageNodeCID, 0)
err := tx.Select(&storageNodeCIDs, pgStr, args...)
return storageNodeCIDs, err
}
// ADD IF LEAF ONLY!!
pkg/ipfs/screener.go
+3 -4
View File
@@ -19,12 +19,11 @@ package ipfs
import (
"bytes"
"github.com/vulcanize/vulcanizedb/pkg/config"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/rlp"
"github.com/vulcanize/vulcanizedb/pkg/config"
)
// ResponseScreener is the inteface used to screen eth data and package appropriate data into a response payload
pkg/ipfs/service.go
+20 -14
View File
@@ -121,7 +121,8 @@ func (sap *Service) APIs() []rpc.API {
}
// SyncAndPublish is the backend processing loop which streams data from geth, converts it to iplds, publishes them to ipfs, and indexes their cids
// It then forwards the data to the Serve() loop which filters and sends relevant data to client subscriptions
// This continues on no matter if or how many subscribers there are, it then forwards the data to the ScreenAndServe() loop
// which filters and sends relevant data to client subscriptions, if there are any
func (sap *Service) SyncAndPublish(wg *sync.WaitGroup, forwardPayloadChan chan<- IPLDPayload, forwardQuitchan chan<- bool) error {
sub, err := sap.Streamer.Stream(sap.PayloadChan)
if err != nil {
@@ -173,7 +174,8 @@ func (sap *Service) SyncAndPublish(wg *sync.WaitGroup, forwardPayloadChan chan<-
return nil
}
// ScreenAndServe is the processing loop used to screen data streamed from the state diffing eth node and send the appropriate data to a requesting client subscription
// ScreenAndServe is the loop used to screen data streamed from the state diffing eth node
// and send the appropriate portions of it to a requesting client subscription, according to their subscription configuration
func (sap *Service) ScreenAndServe(wg *sync.WaitGroup, receivePayloadChan <-chan IPLDPayload, receiveQuitchan <-chan bool) {
wg.Add(1)
go func() {
@@ -206,32 +208,32 @@ func (sap *Service) processResponse(payload IPLDPayload) error {
// Subscribe is used by the API to subscribe to the service loop
func (sap *Service) Subscribe(id rpc.ID, sub chan<- ResponsePayload, quitChan chan<- bool, streamFilters *config.Subscription) {
log.Info("Subscribing to the statediff service")
log.Info("Subscribing to the seed node service")
sap.Lock()
sap.Subscriptions[id] = Subscription{
PayloadChan: sub,
QuitChan: quitChan,
StreamFilters: streamFilters,
}
sap.Unlock()
// If the subscription requests a backfill, use the Postgres index to lookup and retrieve historical data
// Otherwise we only filter new data as it is streamed in from the state diffing geth node
if streamFilters.BackFill {
if streamFilters.BackFill || streamFilters.BackFillOnly {
log.Debug("back-filling data for id", id)
// Retrieve cached CIDs relevant to this subscriber
cids, err := sap.Retriever.RetrieveCIDs(*streamFilters)
cidWrappers, err := sap.Retriever.RetrieveCIDs(*streamFilters)
if err != nil {
log.Error(err)
sap.serve(id, ResponsePayload{
Err: err,
ErrMsg: "CID retrieval error: " + err.Error(),
})
return
}
for _, cid := range cids {
blocksWrapper, err := sap.Fetcher.FetchCIDs(cid)
for _, cidWrapper := range cidWrappers {
blocksWrapper, err := sap.Fetcher.FetchCIDs(cidWrapper)
if err != nil {
log.Error(err)
sap.serve(id, ResponsePayload{
Err: err,
ErrMsg: "IPLD fetching error: " + err.Error(),
})
return
}
@@ -239,18 +241,22 @@ func (sap *Service) Subscribe(id rpc.ID, sub chan<- ResponsePayload, quitChan ch
if err != nil {
log.Error(err)
sap.serve(id, ResponsePayload{
Err: err,
ErrMsg: "IPLD resolving error: " + err.Error(),
})
return
}
sap.serve(id, *backFillIplds)
}
if streamFilters.BackFillOnly {
delete(sap.Subscriptions, id)
}
}
sap.Unlock()
}
// Unsubscribe is used to unsubscribe to the StateDiffingService loop
func (sap *Service) Unsubscribe(id rpc.ID) error {
log.Info("Unsubscribing from the statediff service")
log.Info("Unsubscribing from the seed node service")
sap.Lock()
_, ok := sap.Subscriptions[id]
if !ok {
@@ -263,7 +269,7 @@ func (sap *Service) Unsubscribe(id rpc.ID) error {
// Start is used to begin the service
func (sap *Service) Start(*p2p.Server) error {
log.Info("Starting statediff service")
log.Info("Starting seed node service")
wg := new(sync.WaitGroup)
payloadChan := make(chan IPLDPayload, payloadChanBufferSize)
quitChan := make(chan bool, 1)
@@ -276,7 +282,7 @@ func (sap *Service) Start(*p2p.Server) error {
// Stop is used to close down the service
func (sap *Service) Stop() error {
log.Info("Stopping statediff service")
log.Info("Stopping seed node service")
close(sap.QuitChan)
return nil
}
pkg/ipfs/service_test.go
+7 -8
View File
@@ -26,8 +26,7 @@ import (
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/ipfs"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers/mocks"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
)
var _ = Describe("Service", func() {
@@ -41,18 +40,18 @@ var _ = Describe("Service", func() {
ReturnErr: nil,
}
mockPublisher := &mocks.IPLDPublisher{
ReturnCIDPayload: &helpers.MockCIDPayload,
ReturnCIDPayload: &mocks.MockCIDPayload,
ReturnErr: nil,
}
mockStreamer := &mocks.StateDiffStreamer{
ReturnSub: &rpc.ClientSubscription{},
StreamPayloads: []statediff.Payload{
helpers.MockStatediffPayload,
mocks.MockStatediffPayload,
},
ReturnErr: nil,
}
mockConverter := &mocks.PayloadConverter{
ReturnIPLDPayload: &helpers.MockIPLDPayload,
ReturnIPLDPayload: &mocks.MockIPLDPayload,
ReturnErr: nil,
}
processor := &ipfs.Service{
@@ -68,9 +67,9 @@ var _ = Describe("Service", func() {
time.Sleep(2 * time.Second)
quitChan <- true
wg.Wait()
Expect(mockConverter.PassedStatediffPayload).To(Equal(helpers.MockStatediffPayload))
Expect(mockCidRepo.PassedCIDPayload).To(Equal(&helpers.MockCIDPayload))
Expect(mockPublisher.PassedIPLDPayload).To(Equal(&helpers.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))
})
})
pkg/ipfs/streamer_test.go
+3 -4
View File
@@ -22,8 +22,7 @@ import (
. "github.com/onsi/ginkgo"
. "github.com/onsi/gomega"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/helpers/mocks"
"github.com/vulcanize/vulcanizedb/pkg/ipfs/mocks"
)
var _ = Describe("Streamer", func() {
@@ -32,7 +31,7 @@ var _ = Describe("Streamer", func() {
mockStreamer := mocks.StateDiffStreamer{}
mockStreamer.ReturnSub = &rpc.ClientSubscription{}
mockStreamer.StreamPayloads = []statediff.Payload{
helpers.MockStatediffPayload,
mocks.MockStatediffPayload,
}
payloadChan := make(chan statediff.Payload, 1)
sub, err := mockStreamer.Stream(payloadChan)
@@ -40,7 +39,7 @@ var _ = Describe("Streamer", func() {
Expect(sub).To(Equal(&rpc.ClientSubscription{}))
Expect(mockStreamer.PassedPayloadChan).To(Equal(payloadChan))
streamedPayload := <-payloadChan
Expect(streamedPayload).To(Equal(helpers.MockStatediffPayload))
Expect(streamedPayload).To(Equal(mocks.MockStatediffPayload))
})
})
})
pkg/ipfs/types.go
+1 -1
View File
@@ -43,7 +43,7 @@ type ResponsePayload struct {
ReceiptsRlp [][]byte `json:"receiptsRlp"`
StateNodesRlp map[common.Hash][]byte `json:"stateNodesRlp"`
StorageNodesRlp map[common.Hash]map[common.Hash][]byte `json:"storageNodesRlp"`
Err error `json:"error"`
ErrMsg string `json:"errMsg"`
encoded []byte
err error