From c8742b1f1498ffe6f618451ada3da3b3ecf2509f Mon Sep 17 00:00:00 2001
From: Quentin Dufour <quentin@deuxfleurs.fr>
Date: Fri, 22 May 2020 19:21:11 +0200
Subject: Connecting elements

---
 src/diplonat.rs  | 18 ++++++++++++------
 src/igd_actor.rs | 26 ++++++++++++++------------
 src/main.rs      |  1 +
 3 files changed, 27 insertions(+), 18 deletions(-)

(limited to 'src')

diff --git a/src/diplonat.rs b/src/diplonat.rs
index a4cb787..565c567 100644
--- a/src/diplonat.rs
+++ b/src/diplonat.rs
@@ -1,28 +1,34 @@
 use anyhow::Result;
-use futures::future::try_join_all;
 use log::*;
+use tokio::try_join;
 use crate::consul_actor::ConsulActor;
+use crate::igd_actor::IgdActor;
 use crate::environment::Environment;
 
 pub struct Diplonat {
-  consul: ConsulActor
+  consul: ConsulActor,
+  igd: IgdActor
 }
 
 impl Diplonat {
   pub async fn new() -> Result<Self> {
     let env = Environment::new()?;
+    let ca = ConsulActor::new(&env.consul_url, &env.consul_node_name);
+    let ia = IgdActor::new(&ca.rx_open_ports).await?;
 
     let ctx = Self {
-      consul: ConsulActor::new(&env.consul_url, &env.consul_node_name)
+      consul: ca,
+      igd: ia
     };
 
     return Ok(ctx);
   }
 
   pub async fn listen(&mut self) -> Result<()> {
-    try_join_all(vec![
-      self.consul.listen()
-    ]).await?;
+    try_join!(
+      self.consul.listen(),
+      self.igd.listen()
+    )?;
 
     return Ok(());
   }
diff --git a/src/igd_actor.rs b/src/igd_actor.rs
index 6624ab3..4e2f4b6 100644
--- a/src/igd_actor.rs
+++ b/src/igd_actor.rs
@@ -1,31 +1,33 @@
 use igd::aio::*;
 use log::*;
-use tokio::sync::broadcast;
 use anyhow::{Result, Context};
-use std::cell::Cell;
+use tokio::sync::watch;
+use crate::messages;
 
-use crate::diplonat::*;
-use crate::node_state::*;
-
-pub struct IgdAdapter<'a> {
-  state: &'a Cell<NodeState>,
+pub struct IgdActor {
+  rx_ports: watch::Receiver<messages::PublicExposedPorts>,
   gateway: Gateway,
 }
-impl<'a> IgdAdapter<'a> {
-  pub async fn new(ns: &'a Cell<NodeState>, send: &broadcast::Sender<()>) -> Result<IgdAdapter<'a>> {
+
+impl IgdActor {
+  pub async fn new(rxp: &watch::Receiver<messages::PublicExposedPorts>) -> Result<Self> {
     let gw = search_gateway(Default::default())
               .await
               .context("Failed to find gateway")?;
     info!("Gateway: {}", gw);
 
     let ctx = Self { 
-      state: ns,
-      gateway: gw 
+      gateway: gw,
+      rx_ports: rxp.clone()
     };
+
     return Ok(ctx);
   }
 
-  fn run(&self) -> Result<()> {
+  pub async fn listen(&mut self) -> Result<()> {
+    while let Some(ports) = self.rx_ports.recv().await {
+      println!("received = {:#?}", ports);
+    }
     return Ok(());
   }
 }
diff --git a/src/main.rs b/src/main.rs
index 7f985b1..4df32c8 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -2,6 +2,7 @@ mod messages;
 mod environment;
 mod consul;
 mod consul_actor;
+mod igd_actor;
 mod diplonat;
 
 //use std::net::SocketAddrV4;
-- 
cgit v1.2.3