Skip to content

Commit 04b1a21

Browse files
thmzltthmzlt
authored
[hermes] Pass Wormhole arguments from command line or env. vars (#769)
* Format p2p.go * Pass Wormhole arguments from command line or env. vars * Remove forget calls and let memory be freed (also remove confusing comment) * Use proper types on command line arguments
1 parent 9fea461 commit 04b1a21

File tree

4 files changed

+199
-136
lines changed

4 files changed

+199
-136
lines changed

‎hermes/src/config.rs‎

Lines changed: 15 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -24,13 +24,22 @@ pub enum Options {
2424
#[structopt(long)]
2525
id_secp256k1: Option<PathBuf>,
2626

27-
/// Multiaddress for a Wormhole bootstrap peer.
28-
#[structopt(long)]
29-
wormhole_peer: Option<String>,
27+
/// Network ID for Wormhole
28+
#[structopt(long, env = "WORMHOLE_NETWORK_ID")]
29+
wh_network_id: String,
3030

31-
/// Multiaddress to bind Wormhole P2P to.
32-
#[structopt(long)]
33-
wormhole_addr: Option<Multiaddr>,
31+
/// Multiaddresses for Wormhole bootstrap peers (separated by comma).
32+
#[structopt(long, use_delimiter = true, env = "WORMHOLE_BOOTSTRAP_ADDRS")]
33+
wh_bootstrap_addrs: Vec<Multiaddr>,
34+
35+
/// Multiaddresses to bind Wormhole P2P to (separated by comma)
36+
#[structopt(
37+
long,
38+
use_delimiter = true,
39+
default_value = "/ip4/0.0.0.0/udp/30910/quic,/ip6/::/udp/30910/quic",
40+
env = "WORMHOLE_LISTEN_ADDRS"
41+
)]
42+
wh_listen_addrs: Vec<Multiaddr>,
3443

3544
/// The address to bind the RPC server to.
3645
#[structopt(long, default_value = "127.0.0.1:33999")]

‎hermes/src/main.rs‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -52,8 +52,9 @@ async fn init(_update_channel: Receiver<AccountUpdate>) -> Result<()> {
5252
config::Options::Run {
5353
id: _,
5454
id_secp256k1: _,
55-
wormhole_addr: _,
56-
wormhole_peer: _,
55+
wh_network_id,
56+
wh_bootstrap_addrs,
57+
wh_listen_addrs,
5758
rpc_addr,
5859
p2p_addr,
5960
p2p_peer: _,
@@ -62,7 +63,13 @@ async fn init(_update_channel: Receiver<AccountUpdate>) -> Result<()> {
6263

6364
// Spawn the P2P layer.
6465
log::info!("Starting P2P server on {}", p2p_addr);
65-
network::p2p::spawn(handle_message).await?;
66+
network::p2p::spawn(
67+
handle_message,
68+
wh_network_id.to_string(),
69+
wh_bootstrap_addrs,
70+
wh_listen_addrs,
71+
)
72+
.await?;
6673

6774
// Spawn the RPC server.
6875
log::info!("Starting RPC server on {}", rpc_addr);

‎hermes/src/network/p2p.go‎

Lines changed: 116 additions & 117 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
package main
99

1010
// #include <stdlib.h>
11+
// #include <string.h>
1112
//
1213
// // A structure containing Wormhole VAA observations. This must match on both
1314
// // the Go and Rust side.
@@ -27,6 +28,7 @@ import "C"
2728
import (
2829
"context"
2930
"fmt"
31+
"strings"
3032

3133
"github.com/libp2p/go-libp2p"
3234
"github.com/libp2p/go-libp2p/core/crypto"
@@ -45,123 +47,120 @@ import (
4547
)
4648

4749
//export RegisterObservationCallback
48-
func RegisterObservationCallback(f C.callback_t) {
49-
go func() {
50-
ctx := context.Background()
51-
52-
// Setup base network configuration.
53-
networkID := "/wormhole/mainnet/2"
54-
priv, _, err := crypto.GenerateKeyPair(crypto.Ed25519, -1)
55-
bootstrapPeers := []string{
56-
"/dns4/wormhole-mainnet-v2-bootstrap.certus.one/udp/8999/quic/p2p/12D3KooWQp644DK27fd3d4Km3jr7gHiuJJ5ZGmy8hH4py7fP4FP7",
57-
}
58-
59-
// Setup libp2p Connection Manager.
60-
mgr, err := connmgr.NewConnManager(
61-
100,
62-
400,
63-
connmgr.WithGracePeriod(0),
64-
)
65-
66-
if err != nil {
67-
err := fmt.Errorf("Failed to create connection manager: %w", err)
68-
fmt.Println(err)
69-
return
70-
}
71-
72-
// Setup libp2p Reactor.
73-
h, err := libp2p.New(
74-
libp2p.Identity(priv),
75-
libp2p.ListenAddrStrings(
76-
"/ip4/0.0.0.0/udp/30910/quic",
77-
"/ip6/::/udp/30910/quic",
78-
),
79-
libp2p.Security(libp2ptls.ID, libp2ptls.New),
80-
libp2p.Transport(libp2pquic.NewTransport),
81-
libp2p.ConnectionManager(mgr),
82-
libp2p.Routing(func(h host.Host) (routing.PeerRouting, error) {
83-
bootstrappers := make([]peer.AddrInfo, 0)
84-
for _, addr := range bootstrapPeers {
85-
ma, err := multiaddr.NewMultiaddr(addr)
86-
if err != nil {
87-
continue
88-
}
89-
90-
pi, err := peer.AddrInfoFromP2pAddr(ma)
91-
if err != nil || pi.ID == h.ID() {
92-
continue
93-
}
94-
95-
bootstrappers = append(bootstrappers, *pi)
96-
}
97-
idht, err := dht.New(ctx, h, dht.Mode(dht.ModeServer),
98-
dht.ProtocolPrefix(protocol.ID("/"+networkID)),
99-
dht.BootstrapPeers(bootstrappers...),
100-
)
101-
return idht, err
102-
}),
103-
)
104-
105-
if err != nil {
106-
err := fmt.Errorf("Failed to create libp2p host: %w", err)
107-
fmt.Println(err)
108-
return
109-
}
110-
111-
topic := fmt.Sprintf("%s/%s", networkID, "broadcast")
112-
ps, err := pubsub.NewGossipSub(ctx, h)
113-
if err != nil {
114-
err := fmt.Errorf("Failed to create Pubsub: %w", err)
115-
fmt.Println(err)
116-
return
117-
}
118-
119-
th, err := ps.Join(topic)
120-
if err != nil {
121-
err := fmt.Errorf("Failed to join topic: %w", err)
122-
fmt.Println(err)
123-
return
124-
}
125-
126-
sub, err := th.Subscribe()
127-
if err != nil {
128-
err := fmt.Errorf("Failed to subscribe topic: %w", err)
129-
fmt.Println(err)
130-
return
131-
}
132-
133-
for {
134-
for {
135-
select {
136-
case <-ctx.Done():
137-
return
138-
default:
139-
envelope, err := sub.Next(ctx)
140-
if err != nil {
141-
err := fmt.Errorf("Failed to receive Pubsub message: %w", err)
142-
fmt.Println(err)
143-
return
144-
}
145-
146-
// Definition for GossipMessage is generated by Protobuf, see `p2p.proto`.
147-
var msg GossipMessage
148-
err = proto.Unmarshal(envelope.Data, &msg)
149-
150-
switch msg.Message.(type) {
151-
case *GossipMessage_SignedObservation:
152-
case *GossipMessage_SignedVaaWithQuorum:
153-
vaaBytes := msg.GetSignedVaaWithQuorum().GetVaa()
154-
cBytes := C.CBytes(vaaBytes)
155-
defer C.free(cBytes)
156-
C.invoke(f, C.observation_t{
157-
vaa: (*C.char)(cBytes),
158-
vaa_len: C.size_t(len(vaaBytes)),
159-
})
160-
}
161-
}
162-
}
163-
}
164-
}()
50+
func RegisterObservationCallback(f C.callback_t, network_id, bootstrap_addrs, listen_addrs *C.char) {
51+
networkID := C.GoString(network_id)
52+
bootstrapAddrs := strings.Split(C.GoString(bootstrap_addrs), ",")
53+
listenAddrs := strings.Split(C.GoString(listen_addrs), ",")
54+
55+
go func() {
56+
ctx := context.Background()
57+
58+
// Setup base network configuration.
59+
priv, _, err := crypto.GenerateKeyPair(crypto.Ed25519, -1)
60+
61+
// Setup libp2p Connection Manager.
62+
mgr, err := connmgr.NewConnManager(
63+
100,
64+
400,
65+
connmgr.WithGracePeriod(0),
66+
)
67+
68+
if err != nil {
69+
err := fmt.Errorf("Failed to create connection manager: %w", err)
70+
fmt.Println(err)
71+
return
72+
}
73+
74+
// Setup libp2p Reactor.
75+
h, err := libp2p.New(
76+
libp2p.Identity(priv),
77+
libp2p.ListenAddrStrings(listenAddrs...),
78+
libp2p.Security(libp2ptls.ID, libp2ptls.New),
79+
libp2p.Transport(libp2pquic.NewTransport),
80+
libp2p.ConnectionManager(mgr),
81+
libp2p.Routing(func(h host.Host) (routing.PeerRouting, error) {
82+
bootstrappers := make([]peer.AddrInfo, 0)
83+
for _, addr := range bootstrapAddrs {
84+
ma, err := multiaddr.NewMultiaddr(addr)
85+
if err != nil {
86+
continue
87+
}
88+
89+
pi, err := peer.AddrInfoFromP2pAddr(ma)
90+
if err != nil || pi.ID == h.ID() {
91+
continue
92+
}
93+
94+
bootstrappers = append(bootstrappers, *pi)
95+
}
96+
idht, err := dht.New(ctx, h, dht.Mode(dht.ModeServer),
97+
dht.ProtocolPrefix(protocol.ID("/"+networkID)),
98+
dht.BootstrapPeers(bootstrappers...),
99+
)
100+
return idht, err
101+
}),
102+
)
103+
104+
if err != nil {
105+
err := fmt.Errorf("Failed to create libp2p host: %w", err)
106+
fmt.Println(err)
107+
return
108+
}
109+
110+
topic := fmt.Sprintf("%s/%s", networkID, "broadcast")
111+
ps, err := pubsub.NewGossipSub(ctx, h)
112+
if err != nil {
113+
err := fmt.Errorf("Failed to create Pubsub: %w", err)
114+
fmt.Println(err)
115+
return
116+
}
117+
118+
th, err := ps.Join(topic)
119+
if err != nil {
120+
err := fmt.Errorf("Failed to join topic: %w", err)
121+
fmt.Println(err)
122+
return
123+
}
124+
125+
sub, err := th.Subscribe()
126+
if err != nil {
127+
err := fmt.Errorf("Failed to subscribe topic: %w", err)
128+
fmt.Println(err)
129+
return
130+
}
131+
132+
for {
133+
for {
134+
select {
135+
case <-ctx.Done():
136+
return
137+
default:
138+
envelope, err := sub.Next(ctx)
139+
if err != nil {
140+
err := fmt.Errorf("Failed to receive Pubsub message: %w", err)
141+
fmt.Println(err)
142+
return
143+
}
144+
145+
// Definition for GossipMessage is generated by Protobuf, see `p2p.proto`.
146+
var msg GossipMessage
147+
err = proto.Unmarshal(envelope.Data, &msg)
148+
149+
switch msg.Message.(type) {
150+
case *GossipMessage_SignedObservation:
151+
case *GossipMessage_SignedVaaWithQuorum:
152+
vaaBytes := msg.GetSignedVaaWithQuorum().GetVaa()
153+
cBytes := C.CBytes(vaaBytes)
154+
defer C.free(cBytes)
155+
C.invoke(f, C.observation_t{
156+
vaa: (*C.char)(cBytes),
157+
vaa_len: C.size_t(len(vaaBytes)),
158+
})
159+
}
160+
}
161+
}
162+
}
163+
}()
165164
}
166165

167166
func main() {

0 commit comments

Comments
 (0)