aggregator should be writing the queue items

app/app.go

@@ -16,7 +16,6 @@ package app
 
 import (
 	"encoding/json"
-	"github.com/bluzelle/curium/x/aggregator"
 	"github.com/bluzelle/curium/x/oracle"
 	"math"
 	"os"
@@ -80,7 +79,7 @@ var (
 		tax.AppModuleBasic{},
 		faucet.AppModuleBasic{},
 		oracle.AppModuleBasic{},
-		aggregator.AppModuleBasic{},
+//		aggregator.AppModuleBasic{},
 	)
 
 	// account permissions
@@ -138,7 +137,7 @@ type CRUDApp struct {
 	taxKeeper      tax.Keeper
 	oracleKeeper   oracle.Keeper
 	faucetKeeper   faucet.Keeper
-	aggKeeper      aggregator.Keeper
+//	aggKeeper      aggregator.Keeper
 
 	// Module Manager
 	mm *module.Manager
@@ -164,7 +163,7 @@ func NewCRUDApp(
 		gov.StoreKey, params.StoreKey, crud.StoreKey,
 		tax.StoreKey,
 		faucet.StoreKey, crud.LeaseKey, crud.OwnerKey,
-		oracle.StoreKey,aggregator.StoreKey,
+		oracle.StoreKey, /*aggregator.StoreKey,*/
 	)
 
 	tkeys := sdk.NewTransientStoreKeys(staking.TStoreKey, params.TStoreKey)
@@ -280,12 +279,12 @@ func NewCRUDApp(
 		nil,
 	)
 
-	app.aggKeeper = aggregator.NewKeeper(
-		app.cdc,
-		app.oracleKeeper,
-		keys[aggregator.StoreKey],
-		nil,
-	)
+	//app.aggKeeper = aggregator.NewKeeper(
+	//	app.cdc,
+	//	app.oracleKeeper,
+	//	keys[aggregator.StoreKey],
+	//	nil,
+	//)
 
 	app.faucetKeeper = faucet.NewKeeper(
 		app.supplyKeeper,
@@ -312,11 +311,11 @@ func NewCRUDApp(
 		slashing.NewAppModule(app.slashingKeeper, app.accountKeeper, app.stakingKeeper),
 		staking.NewAppModule(app.stakingKeeper, app.accountKeeper, app.supplyKeeper),
 		oracle.NewAppModule(app.oracleKeeper),
-		aggregator.NewAppModule(app.aggKeeper),
+//		aggregator.NewAppModule(app.aggKeeper),
 	)
 
 	app.mm.SetOrderBeginBlockers(distr.ModuleName, slashing.ModuleName, oracle.ModuleName)
-	app.mm.SetOrderEndBlockers(gov.ModuleName, staking.ModuleName, aggregator.ModuleName)
+	app.mm.SetOrderEndBlockers(gov.ModuleName, staking.ModuleName, /*aggregator.ModuleName*/)
 
 	// Sets the order of Genesis - Order matters, genutil is to always come last
 	// NOTE: The genutils moodule must occur after staking so that pools are
@@ -332,7 +331,7 @@ func NewCRUDApp(
 		tax.ModuleName,
 		supply.ModuleName,
 		oracle.ModuleName,
-		aggregator.ModuleName,
+//		aggregator.ModuleName,
 		genutil.ModuleName,
 	)
 

x/oracle/aggregators/aggregator_manager.go

@@ -0,0 +1,23 @@
+package aggregator
+
+import (
+	tokenAggregator "github.com/bluzelle/curium/x/oracle/aggregators/token"
+	"github.com/bluzelle/curium/x/oracle/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+)
+
+var aggregators = []Aggregator{
+	tokenAggregator.NewTokenAggregator(),
+}
+
+func NotifyAggregators(ctx sdk.Context,  value types.SourceValue, updateFn func(prefix []byte, key []byte, value interface{})) {
+	for _, aggregator := range aggregators {
+		aggregator.SourceValueUpdated(ctx, value, updateFn)
+	}
+}
+
+
+type Aggregator interface {
+	SourceValueUpdated(ctx sdk.Context, value types.SourceValue, updateFn func(prefix []byte, key []byte, value interface{}))
+}
+

x/oracle/aggregators/token/aggregator.go

@@ -0,0 +1,54 @@
+package aggregators
+
+import (
+	"fmt"
+	"github.com/bluzelle/curium/x/oracle/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"strings"
+)
+
+var queueStoreKey = types.NewOracleStoreKey(0x10)
+
+type TokenAggregator struct {
+}
+
+type AggregatorValue struct {
+	Batch    string
+	Symbol   string
+	InSymbol string
+	Value    sdk.Dec
+	Count    int64
+	Height   int64
+}
+
+type AggregatorQueueItem struct {
+	SourceName string
+	Batch      string
+	Symbol     string
+	InSymbol   string
+	Value      sdk.Dec
+	Height     int64
+	Weight     int64
+}
+
+func NewTokenAggregator() TokenAggregator {
+	return TokenAggregator{}
+}
+
+func (ta TokenAggregator) SourceValueUpdated(ctx sdk.Context, value types.SourceValue, updateFn func(prefix []byte, key []byte, value interface{})) {
+	fmt.Println(ctx, value)
+	parts := strings.Split(value.SourceName, "-")
+	aggQueueItem := AggregatorQueueItem{
+		Batch:      value.Batch,
+		SourceName: value.SourceName,
+		Symbol:     parts[1],
+		InSymbol:   parts[3],
+		Value:      value.Value,
+		Height:     value.Height,
+		Weight:     value.Weight,
+	}
+
+	blockStr := fmt.Sprintf("%020d", aggQueueItem.Height)
+	key := queueStoreKey.MakeKey(blockStr, aggQueueItem.Batch, aggQueueItem.SourceName)
+	updateFn(queueStoreKey.Prefix, key, aggQueueItem)
+}

x/oracle/keeper/keeper_aggregator.go

@@ -0,0 +1,12 @@
+package keeper
+
+import (
+	"fmt"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+)
+
+func (k Keeper) StoreAggregatorValue(ctx sdk.Context, key []byte, value []byte) error {
+	// TODO: finish here
+	fmt.Println(ctx, key, value)
+	return nil
+}

x/oracle/keeper/keeper_value.go

@@ -1,6 +1,7 @@
 package keeper
 
 import (
+	aggregator "github.com/bluzelle/curium/x/oracle/aggregators"
 	"github.com/bluzelle/curium/x/oracle/types"
 	storeIterator "github.com/cosmos/cosmos-sdk/store/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
@@ -47,9 +48,15 @@ func (k Keeper) UpdateSourceValue(ctx sdk.Context, batch string, sourceName stri
 	}
 	store.Set(key, k.cdc.MustMarshalBinaryBare(sourceValue))
 
-	for _, listener := range valueUpdateListeners {
-		listener(ctx, sourceValue)
+	//for _, listener := range valueUpdateListeners {
+	//	listener(ctx, sourceValue)
+	//}
+
+	updateFn := func(prefix []byte, key []byte, value interface{}) {
+		combinedKey := append(prefix, key...)
+		k.GetStore(ctx).Set(combinedKey, k.cdc.MustMarshalBinaryBare(value))
 	}
+	aggregator.NotifyAggregators(ctx, sourceValue, updateFn)
 }
 
 

x/oracle/types/keeper.go

@@ -1,6 +1,8 @@
 package types
 
-import "github.com/cosmos/cosmos-sdk/types"
+import (
+	"github.com/cosmos/cosmos-sdk/types"
+)
 
 type GlobalOracleConfig struct {
 	AdminAddress types.AccAddress