update

database/database.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"sync"
 
 	"github.com/exchain/process/leveldb"
 	"github.com/exchain/process/orderbook"
@@ -15,6 +16,7 @@ import (
 type DexDB struct {
 	db             *leveldb.Database
 	accountObjects map[common.Address]*AccountObject
+	sync.Mutex
 }
 
 // New creates a new state from a given trie.
@@ -41,6 +43,39 @@ func (ddb *DexDB) GetPairs() (pairs [][]string, err error) {
 	return
 }
 
+func (ddb *DexDB) CreatePair(pair []string) error {
+	pairs, err := ddb.GetPairs()
+	if err != nil {
+		return err
+	}
+
+	pairs = append(pairs, pair)
+	data, err := json.Marshal(pairs)
+	if err != nil {
+		return err
+	}
+	return ddb.db.Put([]byte("_pairs"), data)
+}
+
+func (ddb *DexDB) DisablePair(pair []string) error {
+	pairs, err := ddb.GetPairs()
+	if err != nil {
+		return err
+	}
+
+	for i, p := range pairs {
+		if p[0] == pair[0] && p[1] == pair[1] {
+			pairs = append(pairs[:i], pairs[i+1:]...)
+			break
+		}
+	}
+	data, err := json.Marshal(pairs)
+	if err != nil {
+		return err
+	}
+	return ddb.db.Put([]byte("_pairs"), data)
+}
+
 func (ddb *DexDB) GetOrNewAcountObjectByAgent(agent common.Address) (acc *AccountObject, err error) {
 	data, err := ddb.db.Get([]byte(fmt.Sprintf("proxy_%s", agent.Hex())))
 	if err != nil && errors.Is(err, goleveldb.ErrNotFound) {
@@ -51,10 +86,14 @@ func (ddb *DexDB) GetOrNewAcountObjectByAgent(agent common.Address) (acc *Accoun
 }
 
 func (ddb *DexDB) GetOrNewAccountObject(address common.Address) (acc *AccountObject, err error) {
+	ddb.Lock()
 	if acc, ok := ddb.accountObjects[address]; ok {
+		ddb.Unlock()
 		return acc, nil
 	}
 
+	ddb.Unlock()
+
 	// get from db
 	data, err := ddb.db.Get(address.Bytes())
 	if err != nil && errors.Is(err, goleveldb.ErrNotFound) {
@@ -70,12 +109,16 @@ func (ddb *DexDB) GetOrNewAccountObject(address common.Address) (acc *AccountObj
 		if err != nil {
 			return nil, err
 		}
+		ddb.Lock()
 		ddb.accountObjects[address] = acc
+		ddb.Unlock()
 		return acc, nil
 	}
 
 	acc = NewAccountObject(ddb.db, address)
+	ddb.Lock()
 	ddb.accountObjects[address] = acc
+	ddb.Unlock()
 	return acc, nil
 }
 

engine/engine.go

@@ -173,6 +173,8 @@ func (e *Engine) AddToQueue(tx types.Transaction) {
 			},
 		}
 		e.txQueue <- pTx
+	// case *types.CreatePairTx
+	// case *types.DisablePairTx
 	default:
 		panic("not implemented")
 	}

engine/process.go

 package engine
 
 import (
-	"errors"
-
 	"github.com/exchain/process/types"
 
 	"github.com/ethereum/go-ethereum/common"
@@ -77,6 +75,28 @@ outer:
 					SignProxyR: &nebulav1.SignProxyReceipt{},
 				},
 			})
+		case nebulav1.TxType_CreatePairTx:
+			receipts.Receipts = append(receipts.Receipts, &nebulav1.TransactionReceipt{
+				Hash:        wtx.Hash().Bytes(),
+				TxType:      tx.TxType,
+				Success:     true,
+				BlockHeight: header.Height,
+				Timestamp:   header.Timestamp,
+				Content: &nebulav1.TransactionReceipt_CreatePairR{
+					CreatePairR: &nebulav1.CreatePairReceipt{},
+				},
+			})
+		case nebulav1.TxType_DisablePairTx:
+			receipts.Receipts = append(receipts.Receipts, &nebulav1.TransactionReceipt{
+				Hash:        wtx.Hash().Bytes(),
+				TxType:      tx.TxType,
+				Success:     true,
+				BlockHeight: header.Height,
+				Timestamp:   header.Timestamp,
+				Content: &nebulav1.TransactionReceipt_DisablePairR{
+					DisablePairR: &nebulav1.DisablePairReceipt{},
+				},
+			})
 		default:
 			panic("not implemented")
 		}
@@ -111,49 +131,7 @@ func (e *Engine) ProcessTx(header *nebulav1.BlockHeader, txs *nebulav1.Transacti
 			receipt.Content = &nebulav1.TransactionReceipt_DepositR{
 				DepositR: &nebulav1.DepositReceipt{},
 			}
-		case nebulav1.TxType_LimitTx:
-			receipt.Content = &nebulav1.TransactionReceipt_LimitR{
-				LimitR: &nebulav1.LimitOrderReceipt{},
-			}
-		case nebulav1.TxType_WithdrawTx:
-			receipt.Content = &nebulav1.TransactionReceipt_WithdrawR{
-				WithdrawR: &nebulav1.WithdrawReceipt{},
-			}
-		case nebulav1.TxType_CancelTx:
-			receipt.Content = &nebulav1.TransactionReceipt_CancelR{
-				CancelR: &nebulav1.CancelOrderReceipt{},
-			}
-		case nebulav1.TxType_MarketTx:
-			receipt.Content = &nebulav1.TransactionReceipt_MarketR{
-				MarketR: &nebulav1.MarketOrderReceipt{},
-			}
-		case nebulav1.TxType_CreatePairTx:
-			err := e.ProcessCreatePair(tx)
-			if err != nil {
-				receipt.Success = false
-			}
-			receipt.Content = &nebulav1.TransactionReceipt_CreatePairR{
-				CreatePairR: &nebulav1.CreatePairReceipt{},
-			}
-		case nebulav1.TxType_DisablePairTx:
-			err := e.ProcessDisablePair(tx)
-			if err != nil {
-				receipt.Success = false
-			}
-			receipt.Content = &nebulav1.TransactionReceipt_DisablePairR{
-				DisablePairR: &nebulav1.DisablePairReceipt{},
-			}
-		case nebulav1.TxType_ProtocolTx:
-			receipt.Content = &nebulav1.TransactionReceipt_ProtocolR{
-				ProtocolR: &nebulav1.ProtocolTransactionReceipt{},
-			}
-		case nebulav1.TxType_SignProxyTx:
-			receipt.Content = &nebulav1.TransactionReceipt_SignProxyR{
-				SignProxyR: &nebulav1.SignProxyReceipt{},
-			}
 		default:
-			// TODO: return error
-			return receipts, errors.New("not implemented")
 		}
 		receipts.Receipts = append(receipts.Receipts, receipt)
 	}

engine/system.go

 package engine
 
-import nebulav1 "github.com/exchain/go-exchain/exchain/protocol/gen/go/nebula/v1"
+import (
+	"strings"
 
-func (e *Engine) ProcessCreatePair(tx *nebulav1.Transaction) error {
-	panic("not implemented")
+	nebulav1 "github.com/exchain/go-exchain/exchain/protocol/gen/go/nebula/v1"
+)
+
+func (e *Engine) ProcessCreatePair(tx *nebulav1.CreatePairTransaction) error {
+	err := e.db.CreatePair([]string{tx.BaseCoin.Symbol, tx.QuoteCoin.Symbol})
+	return err
 }
 
-func (e *Engine) ProcessDisablePair(tx *nebulav1.Transaction) error {
-	panic("not implemented")
+func (e *Engine) ProcessDisablePair(tx *nebulav1.DisablePairTransaction) error {
+	pair := strings.Split(tx.PairName, "-")
+	err := e.db.DisablePair(pair)
+	return err
 }