Add an extra hop in the path length now that we have fewer threads

[dnsseed-rust] / src / bgp_client.rs

diff --git a/src/bgp_client.rs b/src/bgp_client.rs
index ba3a16418087196a5dd3c335ad6ea3dfc1d391de..60c80cf0ad9dbfda7fe172548868e7960f61c333 100644 (file)
--- a/src/bgp_client.rs
+++ b/src/bgp_client.rs
@@ -22,7 +22,7 @@ use futures::sync::mpsc;
 use crate::printer::{Printer, Stat};
 use crate::timeout_stream::TimeoutStream;
 
-const PATH_SUFFIX_LEN: usize = 2;
+const PATH_SUFFIX_LEN: usize = 3;
 #[derive(Clone)]
 struct Route { // 32 bytes with a path id u32
        path_suffix: [u32; PATH_SUFFIX_LEN],
@@ -31,7 +31,7 @@ struct Route { // 32 bytes with a path id u32
        med: u32,
 }
 #[allow(dead_code)]
-const ROUTE_LEN: usize = 32 - std::mem::size_of::<(u32, Route)>();
+const ROUTE_LEN: usize = 36 - std::mem::size_of::<(u32, Route)>();
 
 // To keep memory tight (and since we dont' need such close alignment), newtype the v4/v6 routing
 // table entries to make sure they are aligned to single bytes.
@@ -142,6 +142,9 @@ impl RoutingTable {
                                }
                        },
                        NLRIEncoding::IP_MPLS(_) => (),
+                       NLRIEncoding::IP_MPLS_WITH_PATH_ID(_) => (),
+                       NLRIEncoding::IP_VPN_MPLS(_) => (),
+                       NLRIEncoding::L2VPN(_) => (),
                };
        }
 
@@ -169,6 +172,9 @@ impl RoutingTable {
                                }
                        },
                        NLRIEncoding::IP_MPLS(_) => (),
+                       NLRIEncoding::IP_MPLS_WITH_PATH_ID(_) => (),
+                       NLRIEncoding::IP_VPN_MPLS(_) => (),
+                       NLRIEncoding::L2VPN(_) => (),
                };
        }
 }
@@ -196,8 +202,8 @@ impl<'a> std::io::Read for BytesDecoder<'a> {
        }
 }
 
-struct MsgCoder<'a>(&'a Printer);
-impl<'a> codec::Decoder for MsgCoder<'a> {
+struct MsgCoder(Option<Capabilities>);
+impl codec::Decoder for MsgCoder {
        type Item = Message;
        type Error = std::io::Error;
 
@@ -206,15 +212,17 @@ impl<'a> codec::Decoder for MsgCoder<'a> {
                        buf: bytes,
                        pos: 0
                };
-               match (Reader {
+               let def_cap = Default::default();
+               let mut reader = Reader {
                        stream: &mut decoder,
-                       capabilities: Capabilities {
-                               FOUR_OCTET_ASN_SUPPORT: true,
-                               EXTENDED_PATH_NLRI_SUPPORT: true,
-                       }
-               }).read() {
+                       capabilities: if let Some(cap) = &self.0 { cap } else { &def_cap },
+               };
+               match reader.read() {
                        Ok((_header, msg)) => {
                                decoder.buf.advance(decoder.pos);
+                               if let Message::Open(ref o) = &msg {
+                                       self.0 = Some(Capabilities::from_parameters(o.parameters.clone()));
+                               }
                                Ok(Some(msg))
                        },
                        Err(e) => match e.kind() {
@@ -224,12 +232,12 @@ impl<'a> codec::Decoder for MsgCoder<'a> {
                }
        }
 }
-impl<'a> codec::Encoder for MsgCoder<'a> {
+impl codec::Encoder for MsgCoder {
        type Item = Message;
        type Error = std::io::Error;
 
        fn encode(&mut self, msg: Message, res: &mut bytes::BytesMut) -> Result<(), std::io::Error> {
-               msg.write(&mut BytesCoder(res))?;
+               msg.encode(&mut BytesCoder(res))?;
                Ok(())
        }
 }
@@ -340,7 +348,7 @@ impl BGPClient {
                                                future::err(())
                                        })
                                }).and_then(move |stream| {
-                                       let (write, read) = Framed::new(stream.0, MsgCoder(printer)).split();
+                                       let (write, read) = Framed::new(stream.0, MsgCoder(None)).split();
                                        let (mut sender, receiver) = mpsc::channel(10); // We never really should send more than 10 messages unless they're dumb
                                        tokio::spawn(write.sink_map_err(|_| { () }).send_all(receiver)
                                                .then(|_| {
@@ -359,7 +367,7 @@ impl BGPClient {
                                                        OpenCapability::AddPath(vec![
                                                                (AFI::IPV4, SAFI::Unicast, AddPathDirection::ReceivePaths),
                                                                (AFI::IPV6, SAFI::Unicast, AddPathDirection::ReceivePaths)]),
-                                               ])]
+                                               ])],
                                        }));
                                        TimeoutStream::new_persistent(read, timeout).for_each(move |bgp_msg| {
                                                if client.shutdown.load(Ordering::Relaxed) {