293 lines
8.0 KiB
Go
293 lines
8.0 KiB
Go
package node
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"io/ioutil"
|
|
"sync"
|
|
"time"
|
|
|
|
consensus "github.com/Secured-Finance/p2p-oracle-node/consensus"
|
|
"github.com/Secured-Finance/p2p-oracle-node/handler"
|
|
"github.com/Secured-Finance/p2p-oracle-node/rpc"
|
|
"github.com/Secured-Finance/p2p-oracle-node/rpcclient"
|
|
"github.com/filecoin-project/go-address"
|
|
lotusTypes "github.com/filecoin-project/lotus/chain/types"
|
|
"github.com/filecoin-project/specs-actors/actors/abi"
|
|
"github.com/libp2p/go-libp2p-core/host"
|
|
peer "github.com/libp2p/go-libp2p-core/peer"
|
|
discovery "github.com/libp2p/go-libp2p-discovery"
|
|
dht "github.com/libp2p/go-libp2p-kad-dht"
|
|
peerstore "github.com/libp2p/go-libp2p-peerstore"
|
|
pubsub "github.com/libp2p/go-libp2p-pubsub"
|
|
"github.com/multiformats/go-multiaddr"
|
|
)
|
|
|
|
var (
|
|
LotusHost = ""
|
|
LotusJWT = ""
|
|
EthWsUrl = "wss://ropsten.infura.io/ws/v3/b9faa807bb814588bfdb3d6e94a37737"
|
|
EthUrl = "https://ropsten.infura.io/v3/b9faa807bb814588bfdb3d6e94a37737"
|
|
)
|
|
|
|
type LotusMessage struct {
|
|
Version int64
|
|
|
|
To address.Address
|
|
From address.Address
|
|
|
|
Nonce uint64
|
|
|
|
Value lotusTypes.BigInt
|
|
|
|
GasPrice lotusTypes.BigInt
|
|
GasLimit int64
|
|
|
|
Method abi.MethodNum
|
|
Params []byte
|
|
}
|
|
|
|
func (node *Node) readSub(subscription *pubsub.Subscription, incomingMessagesChan chan pubsub.Message) {
|
|
ctx := node.GlobalCtx
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
}
|
|
msg, err := subscription.Next(context.Background())
|
|
if err != nil {
|
|
node.Logger.Warn("Error reading from buffer", err)
|
|
return
|
|
}
|
|
|
|
if string(msg.Data) == "" {
|
|
return
|
|
}
|
|
if string(msg.Data) != "\n" {
|
|
addr, err := peer.IDFromBytes(msg.From)
|
|
if err != nil {
|
|
node.Logger.Warn("Error occurred when reading message From field...", err)
|
|
return
|
|
}
|
|
|
|
// This checks if sender address of incoming message is ours. It is need because we get our messages when subscribed to the same topic.
|
|
if addr == node.Host.ID() {
|
|
continue
|
|
}
|
|
incomingMessagesChan <- *msg
|
|
}
|
|
|
|
}
|
|
}
|
|
|
|
// Subscribes to a topic and then get messages ..
|
|
func (node *Node) newTopic(topic string) {
|
|
ctx := node.GlobalCtx
|
|
subscription, err := node.PubSub.Subscribe(topic)
|
|
if err != nil {
|
|
node.Logger.Warn("Error occurred when subscribing to topic", err)
|
|
return
|
|
}
|
|
time.Sleep(3 * time.Second)
|
|
incomingMessages := make(chan pubsub.Message)
|
|
|
|
go node.readSub(subscription, incomingMessages)
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case msg := <-incomingMessages:
|
|
{
|
|
node.handler.HandleIncomingMessage(node.OracleTopic, msg, func(textMessage handler.EventMessage) {
|
|
node.Logger.Info("%s \x1b[32m%s\x1b[0m> ", textMessage.From, textMessage.Body)
|
|
})
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Write messages to subscription (topic)
|
|
// NOTE: we don't need to be subscribed to publish something
|
|
func (node *Node) writeTopic(topic string) {
|
|
ctx := node.GlobalCtx
|
|
// stdReader := bufio.NewReader(os.Stdin)
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
}
|
|
node.Logger.Info("> ")
|
|
message := &handler.BaseMessage{
|
|
Body: &rpcclient.OracleEvent{},
|
|
Flag: handler.FlagGenericMessage,
|
|
}
|
|
|
|
sendData, err := json.Marshal(message)
|
|
if err != nil {
|
|
node.Logger.Warn("Error occurred when marshalling message object")
|
|
continue
|
|
}
|
|
err = node.PubSub.Publish(topic, sendData)
|
|
if err != nil {
|
|
node.Logger.Warn("Error occurred when publishing", err)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (node *Node) getNetworkTopics() {
|
|
// ctx := node.GlobalCtx
|
|
node.handler.RequestNetworkTopics()
|
|
}
|
|
|
|
func (node *Node) startPubSub(ctx context.Context, host host.Host) {
|
|
pb, err := pubsub.NewGossipSub(ctx, host)
|
|
if err != nil {
|
|
node.Logger.Fatal("Error occurred when create PubSub", err)
|
|
}
|
|
|
|
// pb, err := pubsub.NewFloodsubWithProtocols(context.Background(), host, []protocol.ID{protocol.ID(node.Config.ProtocolID)}, pubsub.WithMessageSigning(true), pubsub.WithStrictSignatureVerification(true))
|
|
// if err != nil {
|
|
// node.Logger.Fatal("Error occurred when create PubSub", err)
|
|
// }
|
|
|
|
// Set global PubSub object
|
|
node.PubSub = pb
|
|
|
|
node.handler = handler.NewHandler(pb, node.OracleTopic, host.ID(), node.networkTopics)
|
|
|
|
kademliaDHT, err := dht.New(ctx, host)
|
|
if err != nil {
|
|
node.Logger.Fatal("Failed to set a new DHT:", err)
|
|
}
|
|
|
|
if err = kademliaDHT.Bootstrap(ctx); err != nil {
|
|
node.Logger.Fatal(err)
|
|
}
|
|
|
|
if !node.Config.Bootstrap {
|
|
var wg sync.WaitGroup
|
|
bootstrapMultiaddr, err := multiaddr.NewMultiaddr(node.Config.BootstrapNodeMultiaddr)
|
|
if err != nil {
|
|
node.Logger.Fatal(err)
|
|
}
|
|
peerinfo, _ := peer.AddrInfoFromP2pAddr(bootstrapMultiaddr)
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if err := host.Connect(ctx, *peerinfo); err != nil {
|
|
node.Logger.Fatal(err)
|
|
} else {
|
|
node.Logger.Info("Connection established with bootstrap node:", *peerinfo)
|
|
}
|
|
}()
|
|
wg.Wait()
|
|
}
|
|
|
|
node.Logger.Info("Announcing ourselves...")
|
|
routingDiscovery := discovery.NewRoutingDiscovery(kademliaDHT)
|
|
discovery.Advertise(ctx, routingDiscovery, node.Config.Rendezvous)
|
|
node.Logger.Info("Successfully announced!")
|
|
|
|
// Randezvous string = service tag
|
|
// Disvover all peers with our service (all ms devices)
|
|
node.Logger.Info("Searching for other peers...")
|
|
peerChan, err := routingDiscovery.FindPeers(ctx, node.Config.Rendezvous)
|
|
if err != nil {
|
|
node.Logger.Fatal("Failed to find new peers, exiting...", err)
|
|
}
|
|
|
|
// NOTE: here we use Randezvous string as 'topic' by default .. topic != service tag
|
|
node.OracleTopic = node.Config.Rendezvous
|
|
subscription, err := pb.Subscribe(node.OracleTopic)
|
|
if err != nil {
|
|
node.Logger.Warn("Error occurred when subscribing to topic", err)
|
|
return
|
|
}
|
|
|
|
node.Logger.Info("Waiting for correct set up of PubSub...")
|
|
time.Sleep(3 * time.Second)
|
|
peers := node.Host.Peerstore().Peers()
|
|
|
|
consensus := consensus.NewRaftConsensus()
|
|
node.Consensus = consensus
|
|
node.Consensus.StartConsensus(node.Host, peers)
|
|
|
|
ethereum := rpcclient.NewEthereumClient()
|
|
node.Ethereum = ethereum
|
|
ethereum.Connect(ctx, EthUrl, "rpc")
|
|
ethereum.Connect(ctx, EthWsUrl, "websocket")
|
|
|
|
lotus := rpc.NewLotusClient(LotusHost, LotusJWT)
|
|
node.Lotus = lotus
|
|
incomingEvents := make(chan rpcclient.OracleEvent)
|
|
incomingMessages := make(chan pubsub.Message)
|
|
|
|
go func() {
|
|
node.writeTopic(node.OracleTopic)
|
|
node.GlobalCtxCancel()
|
|
}()
|
|
go node.readSub(subscription, incomingMessages)
|
|
go ethereum.SubscribeOnOracleEvents(ctx, "0x89d3A6151a9E608c51FF70E0F7f78a109949c2c1", incomingEvents)
|
|
go node.getNetworkTopics()
|
|
|
|
MainLoop:
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
break MainLoop
|
|
case msg := <-incomingMessages:
|
|
{
|
|
node.handler.HandleIncomingMessage(node.OracleTopic, msg, func(textMessage handler.EventMessage) {
|
|
node.Logger.Info("%s > \x1b[32m%s\x1b[0m", textMessage.From, textMessage.Body)
|
|
node.Logger.Info("> ")
|
|
response, err := node.Lotus.GetMessage(textMessage.Body.RequestType)
|
|
if err != nil {
|
|
node.Logger.Warn("Failed to get transaction data from lotus node")
|
|
}
|
|
defer response.Body.Close()
|
|
body, err := ioutil.ReadAll(response.Body)
|
|
if err != nil {
|
|
node.Logger.Warn("Failed to read lotus response")
|
|
}
|
|
var lotusMessage = new(LotusMessage)
|
|
if err := json.Unmarshal(body, &lotusMessage); err != nil {
|
|
node.Logger.Warn("Failed to unmarshal to get message request")
|
|
}
|
|
node.Consensus.UpdateConsensus(lotusMessage.Value.String())
|
|
})
|
|
}
|
|
case newPeer := <-peerChan:
|
|
{
|
|
node.Logger.Info("\nFound peer:", newPeer, ", add address to peerstore")
|
|
|
|
// Adding peer addresses to local peerstore
|
|
host.Peerstore().AddAddr(newPeer.ID, newPeer.Addrs[0], peerstore.PermanentAddrTTL)
|
|
// Connect to the peer
|
|
if err := host.Connect(ctx, newPeer); err != nil {
|
|
node.Logger.Warn("Connection failed:", err)
|
|
}
|
|
node.Logger.Info("Connected to:", newPeer)
|
|
node.Logger.Info("> ")
|
|
}
|
|
case event := <-incomingEvents:
|
|
{
|
|
message := &handler.BaseMessage{
|
|
Body: &event,
|
|
Flag: handler.FlagEventMessage,
|
|
From: node.Host.ID(),
|
|
To: "",
|
|
}
|
|
node.handler.SendMessageToServiceTopic(message)
|
|
}
|
|
}
|
|
}
|
|
|
|
if err := host.Close(); err != nil {
|
|
node.Logger.Info("\nClosing host failed:", err)
|
|
}
|
|
node.Logger.Info("\nBye")
|
|
}
|