Merge pull request #1 from CortexFoundation/dev

multiply services switch added

Author: ucwong

README.md

@@ -8,3 +8,5 @@ Make sure config the right cortex RPC service endpoint
 ```
 go run cmd/main.go
 ```
+<img width="1416" alt="image" src="https://github.com/CortexFoundation/robot/assets/22344498/9fea7065-9ffe-4076-a840-5dc218247543">
+

cmd/main.go

@@ -23,6 +23,7 @@ func main() {
 	if m, err := robot.New(cfg, true, false, false, nil); err != nil {
 		panic(err)
 	} else {
+		m.SwitchService(robot.SRV_PRINT)
 		if err := m.Start(); err != nil {
 			log.Error("start failed", "err", err)
 			panic(err)

doc.go

@@ -0,0 +1,22 @@
+// Copyright 2023 The CortexTheseus Authors
+// This file is part of the CortexTheseus library.
+//
+// The CortexTheseus library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The CortexTheseus library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the CortexTheseus library. If not, see <http://www.gnu.org/licenses/>.
+
+package robot
+
+const (
+	SRV_MODEL = 0
+	SRV_PRINT = 99
+)

monitor.go

@@ -83,6 +83,8 @@ type Monitor struct {
 	lock sync.RWMutex
 
 	callback chan any
+
+	srv atomic.Int32
 }
 
 // NewMonitor creates a new instance of monitor.
@@ -116,6 +118,7 @@ func New(flag *params.Config, cache, compress, listen bool, callback chan any) (
 		//start: mclock.Now(),
 	}
 
+	// TODO https://github.com/ucwong/golang-kv
 	if fs_, err := backend.NewChainDB(flag); err != nil {
 		log.Error("file storage failed", "err", err)
 		return nil, err
@@ -137,6 +140,8 @@ func New(flag *params.Config, cache, compress, listen bool, callback chan any) (
 
 	m.mode = flag.Mode
 
+	m.srv.Store(SRV_MODEL)
+
 	/*torrents, _ := fs.initTorrents()
 	if m.mode != params.LAZY {
 		for k, v := range torrents {
@@ -222,7 +227,7 @@ func (m *Monitor) indexCheck() error {
 				//	return err
 				//}
 			}
-			log.Warn("Fs storage is reloading ...", "name", m.ckp.Name, "number", checkpoint.TfsCheckPoint, "version", common.BytesToHash(version), "checkpoint", checkpoint.TfsRoot, "blocks", len(m.fs.Blocks()), "files", len(m.fs.Files()), "txs", m.fs.Txs(), "lastNumber", m.lastNumber.Load(), "last in db", m.fs.LastListenBlockNumber())
+			log.Warn("Fs storage is reloading ...", "name", m.ckp.Name, "number", checkpoint.TfsCheckPoint, "version", common.BytesToHash(version), "checkpoint", checkpoint.TfsRoot, "blocks", len(m.fs.Blocks()), "files", len(m.fs.Files()), "txs", m.fs.Txs(), "lastNumber", m.lastNumber.Load(), "last", m.fs.LastListenBlockNumber())
 		} else {
 			log.Info("Fs storage version check passed", "name", m.ckp.Name, "number", checkpoint.TfsCheckPoint, "version", common.BytesToHash(version), "blocks", len(m.fs.Blocks()), "files", len(m.fs.Files()), "txs", m.fs.Txs())
 		}
@@ -646,7 +651,7 @@ func (m *Monitor) syncLatestBlock() {
 						elapsed := time.Duration(mclock.Now()) - time.Duration(m.start)
 						log.Info("Finish sync, listener will be paused", "current", m.currentNumber.Load(), "elapsed", common.PrettyDuration(elapsed), "progress", progress, "end", end, "last", m.lastNumber.Load())
 						//return
-						timer.Reset(time.Millisecond * 1000 * 180)
+						timer.Reset(time.Millisecond * 1000 * 60)
 						end = false
 						continue
 					}
@@ -682,6 +687,10 @@ func (m *Monitor) currentBlock() (uint64, error) {
 }
 
 func (m *Monitor) skip(i uint64) bool {
+	if m.srv.Load() != SRV_MODEL {
+		return false
+	}
+
 	if len(m.ckp.Skips) == 0 || i > m.ckp.Skips[len(m.ckp.Skips)-1].To || i < m.ckp.Skips[0].From {
 		return false
 	}
@@ -753,6 +762,8 @@ func (m *Monitor) syncLastBlock() uint64 {
 				m.lastNumber.Store(i - 1)
 				return 0
 			}
+
+			// batch blocks operation according service category
 			for _, rpcBlock := range blocks {
 				if err := m.solve(rpcBlock); err != nil {
 					m.lastNumber.Store(i - 1)
@@ -808,7 +819,48 @@ func (m *Monitor) syncLastBlock() uint64 {
 	return uint64(maxNumber - minNumber)
 }
 
+// solve block from node
 func (m *Monitor) solve(block *types.Block) error {
+	switch m.srv.Load() {
+	case SRV_MODEL:
+		return m.forModelService(block)
+	//case 1:
+	//	return m.forExplorerService(block) // others service, explorer, exchange, zkp, nft, etc.
+	//case 2:
+	//	return m.forExchangeService(block)
+	case SRV_PRINT:
+		return m.forPrintService(block)
+	default:
+		return errors.New("no block operation service found")
+	}
+}
+
+func (m *Monitor) SwitchService(srv int) error {
+	m.srv.Store(int32(srv))
+	return nil
+}
+
+// only for examples
+func (m *Monitor) forExplorerService(block *types.Block) error {
+	return errors.New("not support")
+}
+
+func (m *Monitor) forExchangeService(block *types.Block) error {
+	return errors.New("not support")
+}
+
+func (m *Monitor) forPrintService(block *types.Block) error {
+	log.Info("Block print", "num", block.Number, "hash", block.Hash, "txs", len(block.Txs))
+	if len(block.Txs) > 0 {
+		for _, t := range block.Txs {
+			log.Info("Tx print", "hash", t.Hash, "amount", t.Amount, "gas", t.GasLimit, "receipt", t.Recipient, "payload", t.Payload)
+		}
+	}
+	m.fs.Anchor(block.Number)
+	return nil
+}
+
+func (m *Monitor) forModelService(block *types.Block) error {
 	i := block.Number
 	if i%65536 == 0 {
 		defer func() {
@@ -836,7 +888,6 @@ func (m *Monitor) solve(block *types.Block) error {
 			log.Debug("Seal fs record", "number", i, "record", record, "root", m.fs.Root().Hex(), "blocks", len(m.fs.Blocks()), "txs", m.fs.Txs(), "files", len(m.fs.Files()), "ckp", m.fs.CheckPoint())
 		} else {
 			if m.fs.LastListenBlockNumber() < i {
-				//m.fs.LastListenBlockNumber = i
 				m.fs.Anchor(i)
 			}