config: refactor to send more specific messages to peer_server on change

4 files changed, 95 insertions, 58 deletions

diff --git a/Cargo.lock b/Cargo.lock
index 4d58c95..d1ee78d 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -197,8 +197,8 @@ dependencies = [
  "itertools 0.7.8 (registry+https://github.com/rust-lang/crates.io-index)",
  "itertools-num 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde 1.0.43 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde_derive 1.0.43 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde_derive 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 1.0.17 (registry+https://github.com/rust-lang/crates.io-index)",
  "simplelog 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
@@ -438,7 +438,7 @@ dependencies = [
  "pest_derive 1.0.7 (registry+https://github.com/rust-lang/crates.io-index)",
  "quick-error 1.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "regex 0.2.11 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde 1.0.43 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)",
  "serde_json 1.0.17 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
@@ -944,12 +944,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
 name = "serde"
-version = "1.0.43"
+version = "1.0.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 
 [[package]]
 name = "serde_derive"
-version = "1.0.43"
+version = "1.0.44"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "proc-macro2 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -974,7 +974,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "dtoa 0.4.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "itoa 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "serde 1.0.43 (registry+https://github.com/rust-lang/crates.io-index)",
+ "serde 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1207,17 +1207,18 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "0.1.5"
+version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
  "mio 0.6.14 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-executor 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-fs 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-io 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-reactor 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-tcp 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio-threadpool 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio-timer 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-threadpool 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-timer 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-udp 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
@@ -1232,11 +1233,11 @@ dependencies = [
  "log 0.4.1 (registry+https://github.com/rust-lang/crates.io-index)",
  "mio 0.6.14 (registry+https://github.com/rust-lang/crates.io-index)",
  "scoped-tls 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-executor 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-io 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-reactor 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio-timer 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-timer 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)",
 ]
 
 [[package]]
@@ -1248,6 +1249,16 @@ dependencies = [
 ]
 
 [[package]]
+name = "tokio-fs"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+dependencies = [
+ "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-io 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio-threadpool 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)",
+]
+
+[[package]]
 name = "tokio-io"
 version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
@@ -1299,7 +1310,7 @@ dependencies = [
 
 [[package]]
 name = "tokio-threadpool"
-version = "0.1.2"
+version = "0.1.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "crossbeam-deque 0.3.0 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1312,7 +1323,7 @@ dependencies = [
 
 [[package]]
 name = "tokio-timer"
-version = "0.2.1"
+version = "0.2.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 dependencies = [
  "futures 0.1.21 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1492,7 +1503,7 @@ dependencies = [
  "structopt 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)",
  "structopt-derive 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "subtle 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)",
- "tokio 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
+ "tokio 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-core 0.1.17 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-io 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)",
  "tokio-signal 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)",
@@ -1634,8 +1645,8 @@ dependencies = [
 "checksum scopeguard 0.3.3 (registry+https://github.com/rust-lang/crates.io-index)" = "94258f53601af11e6a49f722422f6e3425c52b06245a5cf9bc09908b174f5e27"
 "checksum semver 0.9.0 (registry+https://github.com/rust-lang/crates.io-index)" = "1d7eb9ef2c18661902cc47e535f9bc51b78acd254da71d375c2f6720d9a40403"
 "checksum semver-parser 0.7.0 (registry+https://github.com/rust-lang/crates.io-index)" = "388a1df253eca08550bef6c72392cfe7c30914bf41df5269b68cbd6ff8f570a3"
-"checksum serde 1.0.43 (registry+https://github.com/rust-lang/crates.io-index)" = "0c855d888276f20d140223bd06515e5bf1647fd6d02593cb5792466d9a8ec2d0"
-"checksum serde_derive 1.0.43 (registry+https://github.com/rust-lang/crates.io-index)" = "aa113e5fc4b008a626ba2bbd41330b56c9987d667f79f7b243e5a2d03d91ed1c"
+"checksum serde 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)" = "ce67a48047802238bfc88687272de48fd6d7af256b0097f110e968b0017235a5"
+"checksum serde_derive 1.0.44 (registry+https://github.com/rust-lang/crates.io-index)" = "5d870179775231857959d909fd4ab5e8c4e69d537a1e9698707b2ba2c92d370e"
 "checksum serde_derive_internals 0.23.1 (registry+https://github.com/rust-lang/crates.io-index)" = "9d30c4596450fd7bbda79ef15559683f9a79ac0193ea819db90000d7e1cae794"
 "checksum serde_json 1.0.17 (registry+https://github.com/rust-lang/crates.io-index)" = "f3ad6d546e765177cf3dded3c2e424a8040f870083a0e64064746b958ece9cb1"
 "checksum simplelog 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)" = "ce595117de34b75e057b41e99079e43e9fcc4e5ec9c7ba5f2fea55321f0c624e"
@@ -1664,15 +1675,16 @@ dependencies = [
 "checksum thread-scoped 1.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "bcbb6aa301e5d3b0b5ef639c9a9c7e2f1c944f177b460c04dc24c69b1fa2bd99"
 "checksum thread_local 0.3.5 (registry+https://github.com/rust-lang/crates.io-index)" = "279ef31c19ededf577bfd12dfae728040a21f635b06a24cd670ff510edd38963"
 "checksum time 0.1.39 (registry+https://github.com/rust-lang/crates.io-index)" = "a15375f1df02096fb3317256ce2cee6a1f42fc84ea5ad5fc8c421cfe40c73098"
-"checksum tokio 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "be15ef40f675c9fe66e354d74c73f3ed012ca1aa14d65846a33ee48f1ae8d922"
+"checksum tokio 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "7d00555353b013e170ed8bc4e13f648a317d1fd12157dbcae13f7013f6cf29f5"
 "checksum tokio-core 0.1.17 (registry+https://github.com/rust-lang/crates.io-index)" = "aeeffbbb94209023feaef3c196a41cbcdafa06b4a6f893f68779bb5e53796f71"
 "checksum tokio-executor 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "8cac2a7883ff3567e9d66bb09100d09b33d90311feca0206c7ca034bc0c55113"
+"checksum tokio-fs 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "76766830bbf9a2d5bfb50c95350d56a2e79e2c80f675967fff448bc615899708"
 "checksum tokio-io 0.1.6 (registry+https://github.com/rust-lang/crates.io-index)" = "6af9eb326f64b2d6b68438e1953341e00ab3cf54de7e35d92bfc73af8555313a"
 "checksum tokio-reactor 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)" = "b3cedc8e5af5131dc3423ffa4f877cce78ad25259a9a62de0613735a13ebc64b"
 "checksum tokio-signal 0.1.5 (registry+https://github.com/rust-lang/crates.io-index)" = "e8f46863230f9a05cf52d173721ec391b9c5782a2465f593029922b8782b9ffe"
 "checksum tokio-tcp 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "ec9b094851aadd2caf83ba3ad8e8c4ce65a42104f7b94d9e6550023f0407853f"
-"checksum tokio-threadpool 0.1.2 (registry+https://github.com/rust-lang/crates.io-index)" = "bf3d05cdd6a78005e535d2b27c21521bdf91fbb321027a62d8e178929d18966d"
-"checksum tokio-timer 0.2.1 (registry+https://github.com/rust-lang/crates.io-index)" = "29a89e4ad0c8f1e4c9860e605c38c69bfdad3cccd4ea446e58ff588c1c07a397"
+"checksum tokio-threadpool 0.1.3 (registry+https://github.com/rust-lang/crates.io-index)" = "5783254b10c7c84a56f62c74766ef7e5b83d1f13053218c7cab8d3f2c826fa0e"
+"checksum tokio-timer 0.2.3 (registry+https://github.com/rust-lang/crates.io-index)" = "535fed0ccee189f3d48447587697ba3fd234b3dbbb091f0ec4613ddfec0a7c4c"
 "checksum tokio-udp 0.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "137bda266504893ac4774e0ec4c2108f7ccdbcb7ac8dced6305fe9e4e0b5041a"
 "checksum tokio-uds 0.1.7 (registry+https://github.com/rust-lang/crates.io-index)" = "65ae5d255ce739e8537221ed2942e0445f4b3b813daebac1c0050ddaaa3587f9"
 "checksum tokio-utun 0.1.10 (registry+https://github.com/rust-lang/crates.io-index)" = "581cd01e868810f3ca4515605ec2c96ae32964064719b96bccca40dff8240545"
diff --git a/src/interface/config.rs b/src/interface/config.rs
index 488b5bb..882b6cb 100644
--- a/src/interface/config.rs
+++ b/src/interface/config.rs
@@ -3,26 +3,28 @@
 // Dev notes:
 // * Configuration service should use channels to report updates it receives over its interface.
 
+use std::net::SocketAddr;
+use std::{cell::RefCell, iter::Iterator, rc::Rc, mem, str};
+use std::fs::{create_dir, remove_file};
+use std::path::{Path, PathBuf};
+
 use base64;
 use bytes::BytesMut;
-use consts::MAX_PEERS_PER_DEVICE;
 use failure::{Error, err_msg};
-use futures::{Async, Future, Poll, Stream, Sink, future, stream, unsync::mpsc};
-use hex;
+use futures::{Async, Future, Poll, Stream, Sink, future, unsync::mpsc};
+use hex::{self, FromHex};
+use tokio_core::reactor::Handle;
+use tokio_io::{AsyncRead, codec::{Encoder, Decoder}};
+use tokio_uds::UnixListener;
+use x25519_dalek as x25519;
+
+use consts::MAX_PEERS_PER_DEVICE;
 use interface::{SharedState, State};
 use interface::grim_reaper::GrimReaper;
+use interface::peer_server::ChannelMessage;
 use peer::Peer;
-use std::net::SocketAddr;
-use std::{cell::RefCell, iter::Iterator, rc::Rc, mem, str};
-use std::fs::{create_dir, remove_file};
-use std::path::{Path, PathBuf};
-use tokio_core::reactor::Handle;
 use types::PeerInfo;
-use hex::FromHex;
-use x25519_dalek as x25519;
 
-use tokio_io::{AsyncRead, codec::{Encoder, Decoder}};
-use tokio_uds::UnixListener;
 
 #[derive(Debug)]
 pub enum Command {
@@ -140,21 +142,20 @@ pub struct ConfigurationService {
     interface_name: String,
     config_server: Box<Future<Item = (), Error = ()>>,
     reaper: Box<Future<Item = (), Error = ()>>,
-    rx: mpsc::Receiver<UpdateEvent>,
+    rx: mpsc::Receiver<ChannelMessage>
 }
 
 impl ConfigurationService {
     pub fn new(interface_name: &str, state: &SharedState, handle: &Handle) -> Result<Self, Error> {
         let config_path = Self::get_path(interface_name).unwrap();
         let listener    = UnixListener::bind(config_path.clone(), handle).unwrap();
-        let (tx, rx)    = mpsc::channel::<UpdateEvent>(1024);
+        let (tx, rx)    = mpsc::channel::<ChannelMessage>(1024);
 
         // TODO only listen for own socket, verify behavior from `notify` crate
         let reaper = GrimReaper::spawn(handle, &config_path).unwrap();
 
         let config_server = listener.incoming().for_each({
             let handle = handle.clone();
-            let tx = tx.clone();
             let state = state.clone();
             move |(stream, _)| {
                 let (sink, stream) = stream.framed(ConfigurationCodec {}).split();
@@ -169,11 +170,12 @@ impl ConfigurationService {
                         match command {
                             Command::Set(_version, items) => {
                                 for item in &items {
-                                    if Self::handle_update(&mut state, item).is_err() {
-                                        return future::ok("errno=1\nerrno=1\n\n".into());
+                                    match Self::handle_update(&mut state, item) {
+                                        Ok(Some(msg)) => { tx.clone().send(msg).wait().unwrap(); },
+                                        Err(_)        => { return future::ok("errno=1\nerrno=1\n\n".into()); },
+                                        _             => {}
                                     }
                                 }
-                                tx.clone().send_all(stream::iter_ok(items)).wait().unwrap();
                                 future::ok("errno=0\nerrno=0\n\n".into())
                             },
                             Command::Get(_version) => {
@@ -220,13 +222,14 @@ impl ConfigurationService {
         state.router.remove_allowed_ips(&peer.info.allowed_ips);
     }
 
-    pub fn handle_update(state: &mut State, event: &UpdateEvent) -> Result<(), Error> {
+    pub fn handle_update(state: &mut State, event: &UpdateEvent) -> Result<Option<ChannelMessage>, Error> {
         match *event {
             UpdateEvent::PrivateKey(private_key) => {
                 if private_key == [0u8; 32] {
                     state.interface_info.private_key = None;
                     state.interface_info.pub_key     = None;
                     debug!("unset private key");
+                    Ok(Some(ChannelMessage::ClearPrivateKey))
                 } else {
                     let pub_key = x25519::generate_public(&private_key);
                     state.interface_info.private_key = Some(private_key);
@@ -237,16 +240,19 @@ impl ConfigurationService {
                         Self::clear_peer_refs(state, &peer_ref.borrow());
                         debug!("removed self from peers");
                     }
+                    Ok(Some(ChannelMessage::NewPrivateKey))
                 }
             },
             UpdateEvent::ListenPort(port) => {
                 state.interface_info.listen_port = Some(port);
                 debug!("set listen port: {}", port);
+                Ok(Some(ChannelMessage::NewListenPort(port))) // TODO: only notify on listen port *change*
             },
             UpdateEvent::Fwmark(mark) => {
                 state.interface_info.fwmark = Some(mark);
                 debug!("set fwmark: {}", mark);
-            }
+                Ok(Some(ChannelMessage::NewFwmark(mark))) // TODO: only notify on fwmark *change*
+            },
             UpdateEvent::UpdatePeer(ref info, replace_allowed_ips) => {
                 let existing_peer = state.pubkey_map.get(&info.pub_key).cloned();
                 if let Some(peer_ref) = existing_peer {
@@ -258,16 +264,22 @@ impl ConfigurationService {
                     } else {
                         info.allowed_ips.extend_from_slice(&peer.info.allowed_ips);
                     }
+                    let ret = if info.keepalive.is_some() && peer.info.keepalive != info.keepalive {
+                        Some(ChannelMessage::NewPersistentKeepalive(info.keepalive.unwrap()))
+                    } else {
+                        None
+                    };
                     info.endpoint  = info.endpoint.or(peer.info.endpoint);
                     info.keepalive = info.keepalive.or(peer.info.keepalive);
                     info.psk       = info.psk.or(peer.info.psk);
                     state.router.add_allowed_ips(&info.allowed_ips, &peer_ref);
                     peer.info = info;
+                    Ok(ret)
                 } else {
                     if let Some(pub_key) = state.interface_info.pub_key {
                         if pub_key == info.pub_key {
                             debug!("ignoring self-peer add");
-                            return Ok(())
+                            return Ok(None)
                         }
                     }
 
@@ -280,20 +292,22 @@ impl ConfigurationService {
                     let peer_ref = Rc::new(RefCell::new(peer));
                     let _ = state.pubkey_map.insert(info.pub_key, peer_ref.clone());
                     state.router.add_allowed_ips(&info.allowed_ips, &peer_ref);
-                };
+                    Ok(None) // TODO: notify specifically on details of these new peers
+                }
             },
             UpdateEvent::RemoveAllPeers => {
                 state.pubkey_map.clear();
                 state.index_map.clear();
                 state.router.clear();
+                Ok(None)
             },
             UpdateEvent::RemovePeer(pub_key) => {
                 let peer_ref = state.pubkey_map.remove(&pub_key)
                     .ok_or_else(|| err_msg("trying to remove nonexistent peer"))?;
                 Self::clear_peer_refs(state, &peer_ref.borrow());
+                Ok(None)
             },
         }
-        Ok(())
     }
 
     pub fn get_path(interface_name: &str) -> Result<PathBuf, Error> {
@@ -342,7 +356,7 @@ impl ConfigurationService {
 }
 
 impl Stream for ConfigurationService {
-    type Item  = UpdateEvent;
+    type Item  = ChannelMessage;
     type Error = Error;
 
     fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
diff --git a/src/interface/mod.rs b/src/interface/mod.rs
index c670bbe..0d12048 100644
--- a/src/interface/mod.rs
+++ b/src/interface/mod.rs
@@ -99,12 +99,12 @@ impl Interface {
 
         let peer_server   = PeerServer::new(core.handle(), self.state.clone(), utun_tx.clone())?;
         let config_server = ConfigurationService::new(&self.name, &self.state, &core.handle())?;
-        let config_server = config_server.forward(peer_server.config_tx()).map_err(|_|()); // TODO: don't just forward, this is so hacky.
+        let config_server = config_server.forward(peer_server.tx()).map_err(|_|()); // TODO: don't just forward, this is so hacky.
         let utun_stream   = UtunStream::connect(&self.name, &core.handle())?.framed(VecUtunCodec{});
 
         let (utun_writer, utun_reader) = utun_stream.split();
 
-        let utun_read_fut = peer_server.tx()
+        let utun_read_fut = peer_server.tunnel_tx()
             .sink_map_err(|e| -> Error { e.into() })
             .send_all(utun_reader.map_err(|e| -> Error { e.into() }))
             .map_err(|e| { warn!("utun read error: {:?}", e); () });
diff --git a/src/interface/peer_server.rs b/src/interface/peer_server.rs
index 9595deb..aa4f53e 100644
--- a/src/interface/peer_server.rs
+++ b/src/interface/peer_server.rs
@@ -1,7 +1,7 @@
 use consts::{REKEY_TIMEOUT, REKEY_ATTEMPT_TIME, KEEPALIVE_TIMEOUT, STALE_SESSION_TIMEOUT,
              MAX_CONTENT_SIZE, WIPE_AFTER_TIME};
 use cookie;
-use interface::{SharedPeer, SharedState, State, UtunPacket, config};
+use interface::{SharedPeer, SharedState, State, UtunPacket};
 use message::{Message, Initiation, Response, CookieReply, Transport};
 use peer::{Peer, SessionType, SessionTransition};
 use time::Timestamp;
@@ -14,7 +14,15 @@ use rand::{self, Rng};
 use udp::{Endpoint, UdpSocket, PeerServerMessage, UdpChannel};
 use tokio_core::reactor::Handle;
 
-use std::{collections::VecDeque, convert::TryInto, time::Duration};
+use std::{convert::TryInto, time::Duration};
+
+pub enum ChannelMessage {
+    ClearPrivateKey,
+    NewPrivateKey,
+    NewListenPort(u16),
+    NewFwmark(u32),
+    NewPersistentKeepalive(u16),
+}
 
 struct Channel<T> {
     tx: mpsc::Sender<T>,
@@ -36,7 +44,7 @@ pub struct PeerServer {
     udp          : Option<UdpChannel>,
     port         : Option<u16>,
     outgoing     : Channel<UtunPacket>,
-    config       : Channel<config::UpdateEvent>,
+    channel      : Channel<ChannelMessage>,
     timer        : Timer,
     tunnel_tx    : mpsc::Sender<Vec<u8>>,
     cookie       : cookie::Validator,
@@ -51,7 +59,7 @@ impl PeerServer {
             udp      : None,
             port     : None,
             outgoing : mpsc::channel(1024).into(),
-            config   : mpsc::channel(1024).into(),
+            channel  : mpsc::channel(1024).into(),
             cookie   : cookie::Validator::new(&[0u8; 32])
         })
     }
@@ -88,12 +96,12 @@ impl PeerServer {
         Ok(())
     }
 
-    pub fn tx(&self) -> mpsc::Sender<UtunPacket> {
+    pub fn tunnel_tx(&self) -> mpsc::Sender<UtunPacket> {
         self.outgoing.tx.clone()
     }
 
-    pub fn config_tx(&self) -> mpsc::Sender<config::UpdateEvent> {
-        self.config.tx.clone()
+    pub fn tx(&self) -> mpsc::Sender<ChannelMessage> {
+        self.channel.tx.clone()
     }
 
     fn send_to_peer(&self, payload: PeerServerMessage) -> Result<(), Error> {
@@ -376,6 +384,10 @@ impl PeerServer {
             PersistentKeepAlive(peer_ref, our_index) => {
                 let mut peer = peer_ref.borrow_mut();
                 {
+                    if peer.info.keepalive.is_none() {
+                        bail!("no persistent keepalive set for peer.");
+                    }
+
                     let (_, session_type) = peer.find_session(our_index).ok_or_else(|| err_msg("missing session for timer"))?;
                     ensure!(session_type == SessionType::Current, "expired session for persistent keepalive timer");
                 }
@@ -386,7 +398,6 @@ impl PeerServer {
                 if let Some(keepalive) = peer.info.keepalive {
                     self.timer.send_after(Duration::from_secs(u64::from(keepalive)),
                                           PersistentKeepAlive(peer_ref.clone(), our_index));
-
                 }
             },
             Wipe(peer_ref) => {
@@ -413,11 +424,11 @@ impl Future for PeerServer {
     fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
         // Handle config events
         loop {
-            use self::config::UpdateEvent::*;
-            match self.config.rx.poll() {
+            use self::ChannelMessage::*;
+            match self.channel.rx.poll() {
                 Ok(Async::Ready(Some(event))) => {
                     match event {
-                        PrivateKey(_) => {
+                        NewPrivateKey => {
                             let pub_key = self.shared_state.borrow().interface_info.pub_key;
                             if let Some(ref pub_key) = pub_key {
                                 self.cookie = cookie::Validator::new(pub_key);
@@ -429,8 +440,8 @@ impl Future for PeerServer {
                                 self.port = None;
                             }
                         },
-                        ListenPort(_) => self.rebind().unwrap(),
-                        Fwmark(mark) => {
+                        NewListenPort(_) => self.rebind().unwrap(),
+                        NewFwmark(mark) => {
                             if let Some(ref udp) = self.udp {
                                 udp.set_mark(mark).unwrap();
                             }