Move ship_processor.go and types.go into app package.

app/ship_processor.go

@@ -0,0 +1,125 @@
+package app
+
+import (
+	"encoding/hex"
+
+	log "github.com/sirupsen/logrus"
+
+	"eosio-ship-trace-reader/abi"
+	"eosio-ship-trace-reader/transport"
+	"eosio-ship-trace-reader/transport/message"
+	"github.com/eoscanada/eos-go/ship"
+	shipclient "github.com/eosswedenorg-go/antelope-ship-client"
+)
+
+type ShipProcessor struct {
+	ns        transport.Namespace
+	abi       *abi.AbiManager
+	publisher transport.Publisher
+	shClient  *shipclient.ShipClient
+}
+
+func SpawnProccessor(shClient *shipclient.ShipClient, ns transport.Namespace, publisher transport.Publisher, abi *abi.AbiManager) {
+	processor := &ShipProcessor{
+		ns:        ns,
+		abi:       abi,
+		publisher: publisher,
+		shClient:  shClient,
+	}
+
+	// Attach handlers
+	shClient.BlockHandler = processor.processBlock
+	shClient.TraceHandler = processor.processTraces
+}
+
+func (processor *ShipProcessor) queueMessage(channel transport.ChannelInterface, payload []byte) bool {
+	key := processor.ns.NewKey(channel)
+	err := processor.publisher.Publish(key.String(), payload)
+	if err != nil {
+		log.WithError(err).Errorf("Failed to post to channel '%s'", key)
+		return false
+	}
+	return true
+}
+
+func (processor *ShipProcessor) encodeQueue(channel transport.ChannelInterface, v interface{}) bool {
+	if payload, ok := message.Encode(v); ok {
+		if processor.queueMessage(channel, payload) {
+			return true
+		}
+	}
+	return false
+}
+
+func (processor *ShipProcessor) processBlock(block *ship.GetBlocksResultV0) {
+	if block.ThisBlock.BlockNum%100 == 0 {
+		log.Infof("Current: %d, Head: %d\n", block.ThisBlock.BlockNum, block.Head.BlockNum)
+	}
+
+	if block.ThisBlock.BlockNum%10 == 0 {
+		hb := HearthBeat{
+			BlockNum:                 block.ThisBlock.BlockNum,
+			LastIrreversibleBlockNum: block.LastIrreversible.BlockNum,
+			HeadBlockNum:             block.Head.BlockNum,
+		}
+
+		processor.encodeQueue(transport.HeartbeatChannel, hb)
+
+		err := processor.publisher.Flush()
+		if err != nil {
+			log.WithError(err).Error("Failed to send messages")
+		}
+	}
+}
+
+func (processor *ShipProcessor) processTraces(traces []*ship.TransactionTraceV0) {
+	for _, trace := range traces {
+
+		processor.encodeQueue(transport.TransactionChannel, trace)
+
+		// Actions
+		for _, actionTraceVar := range trace.ActionTraces {
+			act_trace := actionTraceVar.Impl.(*ship.ActionTraceV0)
+
+			act := ActionTrace{
+				TxID:     trace.ID,
+				Receiver: act_trace.Receiver,
+				Contract: act_trace.Act.Account,
+				Action:   act_trace.Act.Name,
+				HexData:  hex.EncodeToString(act_trace.Act.Data),
+			}
+
+			ABI, err := processor.abi.GetAbi(act_trace.Act.Account)
+			if err == nil {
+				v, err := abi.DecodeAction(ABI, act_trace.Act.Data, act_trace.Act.Name)
+				if err != nil {
+					log.WithError(err).Warn("Failed to decode action")
+				}
+				act.Data = v
+			} else {
+				log.WithError(err).Errorf("Failed to get abi for contract %s", act_trace.Act.Account)
+			}
+
+			payload, ok := message.Encode(act)
+			if !ok {
+				continue
+			}
+
+			channels := []transport.ChannelInterface{
+				transport.ActionChannel{},
+				transport.ActionChannel{Action: string(act.Action)},
+				transport.ActionChannel{Contract: string(act.Contract)},
+				transport.ActionChannel{Action: string(act.Action), Contract: string(act.Contract)},
+			}
+
+			for _, channel := range channels {
+				processor.queueMessage(channel, payload)
+			}
+		}
+	}
+
+	err := processor.publisher.Flush()
+	if err != nil {
+		log.WithError(err).Error("Failed to send messages")
+	}
+}

app/types.go

@@ -0,0 +1,20 @@
+package app
+
+import (
+	eos "github.com/eoscanada/eos-go"
+)
+
+type HearthBeat struct {
+	BlockNum                 uint32 `json:"blocknum"`
+	HeadBlockNum             uint32 `json:"head_blocknum"`
+	LastIrreversibleBlockNum uint32 `json:"last_irreversible_blocknum"`
+}
+
+type ActionTrace struct {
+	TxID     eos.Checksum256 `json:"tx_id"`
+	Receiver eos.Name        `json:"receiver"`
+	Contract eos.AccountName `json:"contract"`
+	Action   eos.ActionName  `json:"action"`
+	Data     interface{}     `json:"data"`
+	HexData  string          `json:"hex_data"`
+}