Merge branch 'bmp'

1 files changed, 564 insertions, 365 deletions

diff --git a/proto/bmp/bmp.c b/proto/bmp/bmp.c
index f04b59b8..261e9fdd 100644
--- a/proto/bmp/bmp.c
+++ b/proto/bmp/bmp.c
@@ -23,7 +23,7 @@
  *
  * The BMP session is managed by a simple state machine with three states: Idle
  * (!started, !sk), Connect (!started, sk active), and Established (started). It
- * has three events: connect successfull (Connect -> Established), socket error
+ * has three events: connect successful (Connect -> Established), socket error
  * (any -> Idle), and connect timeout (Idle/Connect -> Connect, resetting the
  * TCP socket).
  */
@@ -56,8 +56,25 @@
 #include "nest/iface.h"
 #include "nest/route.h"
 
-// We allow for single instance of BMP protocol
-static struct bmp_proto *g_bmp;
+// List of BMP instances
+static list STATIC_LIST_INIT(bmp_proto_list);
+
+#define HASH_PEER_KEY(n)		n->bgp
+#define HASH_PEER_NEXT(n)		n->next
+#define HASH_PEER_EQ(b1,b2)		b1 == b2
+#define HASH_PEER_FN(b)			ptr_hash(b)
+
+#define BMP_STREAM_KEY_POLICY		0x100
+
+#define HASH_STREAM_KEY(n)		n->bgp, n->key
+#define HASH_STREAM_NEXT(n)		n->next
+#define HASH_STREAM_EQ(b1,k1,b2,k2)	b1 == b2 && k1 == k2
+#define HASH_STREAM_FN(b,k)		ptr_hash(b) ^ u32_hash(k)
+
+#define HASH_TABLE_KEY(n)		n->table
+#define HASH_TABLE_NEXT(n)		n->next
+#define HASH_TABLE_EQ(t1,t2)		t1 == t2
+#define HASH_TABLE_FN(t)		ptr_hash(t)
 
 /* BMP Common Header [RFC 7854 - Section 4.1] */
 enum bmp_version {
@@ -203,18 +220,24 @@ static void bmp_close_socket(struct bmp_proto *p);
 
 static void
 bmp_send_peer_up_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
-  const byte* tx_data, const size_t tx_data_size,
-  const byte* rx_data, const size_t rx_data_size);
+  const byte *tx_data, const size_t tx_data_size,
+  const byte *rx_data, const size_t rx_data_size);
+
+static void bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs);
 
 // Stores necessary any data in list
 struct bmp_data_node {
   node n;
   byte *data;
   size_t data_size;
-};
 
-static void
-bmp_route_monitor_pre_policy_table_in_snapshot(struct channel *C);
+  u32 remote_as;
+  u32 remote_id;
+  ip_addr remote_ip;
+  btime timestamp;
+  bool global_peer;
+  bool policy;
+};
 
 static void
 bmp_common_hdr_serialize(buffer *stream, const enum bmp_message_type type, const u32 data_size)
@@ -229,7 +252,7 @@ bmp_info_tlv_hdr_serialize(buffer *stream, const enum bmp_info_tlv_type type,
   const char *str)
 {
   size_t str_len = strlen(str);
-  str_len = MIN(str_len, 65535);
+  str_len = MIN(str_len, MIB_II_STR_LEN);
 
   bmp_put_u16(stream, type);
   bmp_put_u16(stream, str_len);
@@ -273,7 +296,10 @@ static void
 bmp_fire_tx(void *p_)
 {
   struct bmp_proto *p = p_;
-  byte *buf = p->sk->tbuf;
+
+  if (!p->started)
+    return;
+
   IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
     EMPTY_LIST(p->tx_queue),
     "Called BMP TX event handler when there is not any data to send"
@@ -290,7 +316,7 @@ bmp_fire_tx(void *p_)
     }
 
     size_t data_size = tx_data->data_size;
-    memcpy(buf, tx_data->data, tx_data->data_size);
+    memcpy(p->sk->tbuf, tx_data->data, data_size);
     mb_free(tx_data->data);
     rem_node((node *) tx_data);
     mb_free(tx_data);
@@ -336,19 +362,21 @@ bmp_put_ipa(buffer *stream, const ip_addr addr)
 }
 
 static void
-bmp_set_initial_bgp_hdr(buffer *stream, const u16 msg_size, const u8 msg_type)
+bmp_put_bgp_hdr(buffer *stream, const u8 msg_type, const u16 msg_length)
 {
-  byte marker[BGP_MSG_HDR_MARKER_SIZE];
-  memset(marker, 0xff, BGP_MSG_HDR_MARKER_SIZE);
-  bmp_put_data(stream, marker, BGP_MSG_HDR_MARKER_SIZE);
-  bmp_put_u16(stream, msg_size);
+  bmp_buffer_need(stream, BGP_HEADER_LENGTH);
+
+  memset(stream->pos, 0xff, BGP_HDR_MARKER_LENGTH);
+  stream->pos += BGP_HDR_MARKER_LENGTH;
+
+  bmp_put_u16(stream, msg_length);
   bmp_put_u8(stream, msg_type);
 }
 
 /**
  * bmp_per_peer_hdr_serialize - serializes Per-Peer Header
  *
- * @is_pre_policy: indicate the message reflects the pre-policy Adj-RIB-In
+ * @is_post_policy: indicate the message reflects the post-policy Adj-RIB-In
  * @peer_addr: the remote IP address associated with the TCP session
  * @peer_as: the Autonomous System number of the peer
  * @peer_bgp_id: the BGP Identifier of the peer
@@ -357,7 +385,7 @@ bmp_set_initial_bgp_hdr(buffer *stream, const u16 msg_size, const u8 msg_type)
  */
 static void
 bmp_per_peer_hdr_serialize(buffer *stream, const bool is_global_instance_peer,
-  const bool is_pre_policy, const bool is_as_path_4bytes,
+  const bool is_post_policy, const bool is_as_path_4bytes,
   const ip_addr peer_addr, const u32 peer_as, const u32 peer_bgp_id,
   const u32 ts_sec, const u32 ts_usec)
 {
@@ -368,9 +396,9 @@ bmp_per_peer_hdr_serialize(buffer *stream, const bool is_global_instance_peer,
   const u8 peer_flag_v = ipa_is_ip4(peer_addr)
                            ? BMP_PEER_HDR_FLAG_V_IP4
                            : BMP_PEER_HDR_FLAG_V_IP6;
-  const u8 peer_flag_l = is_pre_policy
-                           ? BMP_PEER_HDR_FLAG_L_PRE_POLICY_ADJ_RIB_IN
-                           : BMP_PEER_HDR_FLAG_L_POST_POLICY_ADJ_RIB_IN;
+  const u8 peer_flag_l = is_post_policy
+                           ? BMP_PEER_HDR_FLAG_L_POST_POLICY_ADJ_RIB_IN
+                           : BMP_PEER_HDR_FLAG_L_PRE_POLICY_ADJ_RIB_IN;
   const u8 peer_flag_a = is_as_path_4bytes
                            ? BMP_PEER_HDR_FLAG_A_AS_PATH_4B
                            : BMP_PEER_HDR_FLAG_A_AS_PATH_2B;
@@ -394,14 +422,17 @@ bmp_per_peer_hdr_serialize(buffer *stream, const bool is_global_instance_peer,
 /* [4.6] Route Monitoring */
 static void
 bmp_route_monitor_msg_serialize(buffer *stream, const bool is_peer_global,
-  const bool table_in_pre_policy, const u32 peer_as, const u32 peer_bgp_id,
+  const bool table_in_post_policy, const u32 peer_as, const u32 peer_bgp_id,
   const bool as4_support, const ip_addr remote_addr, const byte *update_msg,
-  const size_t update_msg_size, u32 ts_sec, u32 ts_usec)
+  const size_t update_msg_size, btime timestamp)
 {
   const size_t data_size = BMP_PER_PEER_HDR_SIZE + update_msg_size;
+  u32 ts_sec = timestamp TO_S;
+  u32 ts_usec = timestamp - (ts_sec S);
+
   bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
   bmp_common_hdr_serialize(stream, BMP_ROUTE_MONITOR, data_size);
-  bmp_per_peer_hdr_serialize(stream, is_peer_global, table_in_pre_policy,
+  bmp_per_peer_hdr_serialize(stream, is_peer_global, table_in_post_policy,
     as4_support, remote_addr, peer_as, peer_bgp_id, ts_sec, ts_usec);
   bmp_put_data(stream, update_msg, update_msg_size);
 }
@@ -410,25 +441,25 @@ static void
 bmp_peer_up_notif_msg_serialize(buffer *stream, const bool is_peer_global,
   const u32 peer_as, const u32 peer_bgp_id, const bool as4_support,
   const ip_addr local_addr, const ip_addr remote_addr, const u16 local_port,
-  const u16 remote_port, const byte *sent_msg, const size_t sent_msg_size,
-  const byte *recv_msg, const size_t recv_msg_size)
+  const u16 remote_port, const byte *sent_msg, const size_t sent_msg_length,
+  const byte *recv_msg, const size_t recv_msg_length)
 {
-  const size_t data_size = BMP_PER_PEER_HDR_SIZE + BMP_PEER_UP_NOTIF_MSG_FIX_SIZE
-                             + sent_msg_size + recv_msg_size;
+  const size_t data_size =
+    BMP_PER_PEER_HDR_SIZE + BMP_PEER_UP_NOTIF_MSG_FIX_SIZE +
+    BGP_HEADER_LENGTH + sent_msg_length + BGP_HEADER_LENGTH + recv_msg_length;
+
   bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + data_size);
   bmp_common_hdr_serialize(stream, BMP_PEER_UP_NOTIF, data_size);
   bmp_per_peer_hdr_serialize(stream, is_peer_global,
-    true /* TODO: Hardcoded pre-policy Adj-RIB-In */, as4_support, remote_addr,
+    false /* TODO: Hardcoded pre-policy Adj-RIB-In */, as4_support, remote_addr,
     peer_as, peer_bgp_id, 0, 0); // 0, 0 - No timestamp provided
   bmp_put_ipa(stream, local_addr);
   bmp_put_u16(stream, local_port);
   bmp_put_u16(stream, remote_port);
-  bmp_set_initial_bgp_hdr(stream, sent_msg_size, PKT_OPEN);
-  const size_t missing_bgp_hdr_size = BGP_MSG_HDR_MARKER_SIZE
-                                        + BGP_MSG_HDR_LENGTH_SIZE
-                                        + BGP_MSG_HDR_TYPE_SIZE;
-  bmp_put_data(stream, sent_msg, sent_msg_size - missing_bgp_hdr_size);
-  bmp_put_data(stream, recv_msg, recv_msg_size);
+  bmp_put_bgp_hdr(stream, PKT_OPEN, BGP_HEADER_LENGTH + sent_msg_length);
+  bmp_put_data(stream, sent_msg, sent_msg_length);
+  bmp_put_bgp_hdr(stream, PKT_OPEN, BGP_HEADER_LENGTH + recv_msg_length);
+  bmp_put_data(stream, recv_msg, recv_msg_length);
 }
 
 static void
@@ -440,57 +471,232 @@ bmp_peer_down_notif_msg_serialize(buffer *stream, const bool is_peer_global,
   bmp_buffer_need(stream, BMP_COMMON_HDR_SIZE + payload_size);
   bmp_common_hdr_serialize(stream, BMP_PEER_DOWN_NOTIF, payload_size);
   bmp_per_peer_hdr_serialize(stream, is_peer_global,
-    true /* TODO: Hardcoded pre-policy adj RIB IN */,  as4_support, remote_addr,
+    false /* TODO: Hardcoded pre-policy adj RIB IN */,  as4_support, remote_addr,
     peer_as, peer_bgp_id, 0, 0); // 0, 0 - No timestamp provided
   bmp_put_data(stream, data, data_size);
 }
 
+
+/*
+ *	BMP tables
+ */
+
+static struct bmp_table *
+bmp_find_table(struct bmp_proto *p, struct rtable *tab)
+{
+  return HASH_FIND(p->table_map, HASH_TABLE, tab);
+}
+
+static struct bmp_table *
+bmp_add_table(struct bmp_proto *p, struct rtable *tab)
+{
+  struct bmp_table *bt = mb_allocz(p->p.pool, sizeof(struct bmp_table));
+  bt->table = tab;
+  rt_lock_table(bt->table);
+
+  HASH_INSERT(p->table_map, HASH_TABLE, bt);
+
+  struct channel_config cc = {
+    .name = "monitor",
+    .channel = &channel_basic,
+    .table = tab->config,
+    .in_filter = FILTER_REJECT,
+    .net_type = tab->addr_type,
+    .ra_mode = RA_ANY,
+    .bmp_hack = 1,
+  };
+
+  bt->channel = proto_add_channel(&p->p, &cc);
+  channel_set_state(bt->channel, CS_UP);
+
+  return bt;
+}
+
 static void
-bmp_peer_map_walk_tx_open_msg_and_send_peer_up_notif(
-  const struct bmp_peer_map_key key, const byte *tx_msg,
-  const size_t tx_msg_size, void *bmp_)
+bmp_remove_table(struct bmp_proto *p, struct bmp_table *bt)
 {
-  struct bmp_proto *p = bmp_;
-  ASSERT(p->started);
+  channel_set_state(bt->channel, CS_FLUSHING);
+  channel_set_state(bt->channel, CS_DOWN);
+  proto_remove_channel(&p->p, bt->channel);
 
-  const struct bmp_peer_map_entry *map_rx_msg = bmp_peer_map_get(&p->peer_open_msg.rx_msg, key);
-  IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    map_rx_msg,
-    "Processing TX BGP OPEN MSG but there is not corresponding received MSG"
-  );
+  HASH_REMOVE(p->table_map, HASH_TABLE, bt);
 
-  const struct bmp_peer_map_entry *map_bgp_proto = bmp_peer_map_get(&p->bgp_peers, key);
-  IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    map_bgp_proto,
-    "There is not BGP proto related with stored TX/RX OPEN MSG"
-  );
+  rt_unlock_table(bt->table);
+  bt->table = NULL;
+
+  mb_free(bt);
+}
+
+static inline struct bmp_table *bmp_get_table(struct bmp_proto *p, struct rtable *tab)
+{ return bmp_find_table(p, tab) ?: bmp_add_table(p, tab); }
+
+static inline void bmp_lock_table(struct bmp_proto *p UNUSED, struct bmp_table *bt)
+{ bt->uc++; }
+
+static inline void bmp_unlock_table(struct bmp_proto *p, struct bmp_table *bt)
+{ bt->uc--; if (!bt->uc) bmp_remove_table(p, bt); }
+
+
+/*
+ *	BMP streams
+ */
+
+static inline u32 bmp_stream_key(u32 afi, bool policy)
+{ return afi ^ (policy ? BMP_STREAM_KEY_POLICY : 0); }
+
+static inline u32 bmp_stream_afi(struct bmp_stream *bs)
+{ return bs->key & ~BMP_STREAM_KEY_POLICY; }
 
-  const struct bgp_proto *bgp;
-  memcpy(&bgp, map_bgp_proto->data.buf, sizeof (bgp));
-  if (bgp->p.proto_state == PS_UP)
+static inline bool bmp_stream_policy(struct bmp_stream *bs)
+{ return !!(bs->key & BMP_STREAM_KEY_POLICY); }
+
+static struct bmp_stream *
+bmp_find_stream(struct bmp_proto *p, const struct bgp_proto *bgp, u32 afi, bool policy)
+{
+  return HASH_FIND(p->stream_map, HASH_STREAM, bgp, bmp_stream_key(afi, policy));
+}
+
+static struct bmp_stream *
+bmp_add_stream(struct bmp_proto *p, struct bmp_peer *bp, u32 afi, bool policy, struct rtable *tab, struct bgp_channel *sender)
+{
+  struct bmp_stream *bs = mb_allocz(p->p.pool, sizeof(struct bmp_stream));
+  bs->bgp = bp->bgp;
+  bs->key = bmp_stream_key(afi, policy);
+
+  add_tail(&bp->streams, &bs->n);
+  HASH_INSERT(p->stream_map, HASH_STREAM, bs);
+
+  bs->table = bmp_get_table(p, tab);
+  bmp_lock_table(p, bs->table);
+
+  bs->sender = sender;
+  bs->sync = false;
+
+  return bs;
+}
+
+static void
+bmp_remove_stream(struct bmp_proto *p, struct bmp_stream *bs)
+{
+  rem_node(&bs->n);
+  HASH_REMOVE(p->stream_map, HASH_STREAM, bs);
+
+  bmp_unlock_table(p, bs->table);
+  bs->table = NULL;
+
+  mb_free(bs);
+}
+
+
+/*
+ *	BMP peers
+ */
+
+static struct bmp_peer *
+bmp_find_peer(struct bmp_proto *p, const struct bgp_proto *bgp)
+{
+  return HASH_FIND(p->peer_map, HASH_PEER, bgp);
+}
+
+static struct bmp_peer *
+bmp_add_peer(struct bmp_proto *p, struct bgp_proto *bgp)
+{
+  struct bmp_peer *bp = mb_allocz(p->p.pool, sizeof(struct bmp_peer));
+  bp->bgp = bgp;
+
+  init_list(&bp->streams);
+
+  HASH_INSERT(p->peer_map, HASH_PEER, bp);
+
+  struct bgp_channel *c;
+  BGP_WALK_CHANNELS(bgp, c)
   {
-    bmp_send_peer_up_notif_msg(p, bgp, tx_msg, tx_msg_size,
-			       map_rx_msg->data.buf, map_rx_msg->data.buf_size);
+    if (p->monitoring_rib.in_pre_policy && c->c.in_table)
+      bmp_add_stream(p, bp, c->afi, false, c->c.in_table, c);
+
+    if (p->monitoring_rib.in_post_policy && c->c.table)
+      bmp_add_stream(p, bp, c->afi, true, c->c.table, c);
   }
+
+  return bp;
 }
 
 static void
-bmp_peer_up(const struct bgp_proto *bgp)
+bmp_remove_peer(struct bmp_proto *p, struct bmp_peer *bp)
 {
-  struct bgp_channel *c;
-  WALK_LIST(c, bgp->p.channels)
+  struct bmp_stream *bs, *bs_next;
+  WALK_LIST_DELSAFE(bs, bs_next, bp->streams)
+    bmp_remove_stream(p, bs);
+
+  HASH_REMOVE(p->peer_map, HASH_PEER, bp);
+
+  mb_free(bp);
+}
+
+static void
+bmp_peer_up_(struct bmp_proto *p, struct bgp_proto *bgp, bool sync,
+	    const byte *tx_open_msg, uint tx_open_length,
+	    const byte *rx_open_msg, uint rx_open_length)
+{
+  if (!p->started)
+    return;
+
+  struct bmp_peer *bp = bmp_find_peer(p, bgp);
+  if (bp)
+    return;
+
+  TRACE(D_STATES, "Peer up for %s", bgp->p.name);
+
+  bp = bmp_add_peer(p, bgp);
+
+  bmp_send_peer_up_notif_msg(p, bgp, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);
+
+  /*
+   * We asssume peer_up() notifications are received before any route
+   * notifications from that peer. Therefore, peers established after BMP
+   * session coould be considered synced with empty RIB.
+   */
+  if (sync)
   {
-    bmp_route_monitor_pre_policy_table_in_snapshot((struct channel *) c);
+    struct bmp_stream *bs;
+    WALK_LIST(bs, bp->streams)
+    {
+      bmp_route_monitor_end_of_rib(p, bs);
+      bs->sync = true;
+    }
   }
 }
 
+void
+bmp_peer_up(struct bgp_proto *bgp,
+	    const byte *tx_open_msg, uint tx_open_length,
+	    const byte *rx_open_msg, uint rx_open_length)
+{
+  struct bmp_proto *p; node *n;
+  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
+    bmp_peer_up_(p, bgp, true, tx_open_msg, tx_open_length, rx_open_msg, rx_open_length);
+}
+
+static void
+bmp_peer_init(struct bmp_proto *p, struct bgp_proto *bgp)
+{
+  struct bgp_conn *conn = bgp->conn;
+
+  if (!conn || (conn->state != BS_ESTABLISHED) ||
+      !conn->local_open_msg || !conn->remote_open_msg)
+    return;
+
+  bmp_peer_up_(p, bgp, false, conn->local_open_msg, conn->local_open_length,
+	       conn->remote_open_msg, conn->remote_open_length);
+}
+
+
+
 static const struct birdsock *
 bmp_get_birdsock(const struct bgp_proto *bgp)
 {
   if (bgp->conn && bgp->conn->sk)
-  {
     return bgp->conn->sk;
-  }
 
   return NULL;
 }
@@ -499,11 +705,8 @@ static const struct birdsock *
 bmp_get_birdsock_ext(const struct bgp_proto *bgp)
 {
   const struct birdsock *sk = bmp_get_birdsock(bgp);
-
   if (sk != NULL)
-  {
     return sk;
-  }
 
   if (bgp->incoming_conn.sk)
   {
@@ -521,9 +724,7 @@ static const struct bgp_caps *
 bmp_get_bgp_remote_caps(const struct bgp_proto *bgp)
 {
   if (bgp->conn && bgp->conn->remote_caps)
-  {
     return bgp->conn->remote_caps;
-  }
 
   return NULL;
 }
@@ -533,9 +734,7 @@ bmp_get_bgp_remote_caps_ext(const struct bgp_proto *bgp)
 {
   const struct bgp_caps *remote_caps = bmp_get_bgp_remote_caps(bgp);
   if (remote_caps != NULL)
-  {
     return remote_caps;
-  }
 
   if (bgp->incoming_conn.remote_caps)
   {
@@ -560,8 +759,8 @@ bmp_is_peer_global_instance(const struct bgp_proto *bgp)
 
 static void
 bmp_send_peer_up_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
-  const byte* tx_data, const size_t tx_data_size,
-  const byte* rx_data, const size_t rx_data_size)
+  const byte *tx_data, const size_t tx_data_size,
+  const byte *rx_data, const size_t rx_data_size)
 {
   ASSERT(p->started);
 
@@ -579,264 +778,98 @@ bmp_send_peer_up_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
     rx_data, rx_data_size);
   bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
   bmp_buffer_free(&payload);
-
-  bmp_peer_up(bgp);
 }
 
-void
-bmp_put_sent_bgp_open_msg(const struct bgp_proto *bgp, const byte* pkt,
-  const size_t pkt_size)
-{
-  struct bmp_proto *p = g_bmp;
-
-  if (!p)
-  {
-    return;
-  }
-
-  struct bmp_peer_map_key key
-    = bmp_peer_map_key_create(bgp->remote_ip, bgp->remote_as);
-  const struct bmp_peer_map_entry *rx_msg
-    = bmp_peer_map_get(&p->peer_open_msg.rx_msg, key);
-
-  bmp_peer_map_insert(&p->peer_open_msg.tx_msg, key, pkt, pkt_size);
-
-  if (!rx_msg)
-    bmp_peer_map_insert(&p->bgp_peers, key, (const byte *) &bgp, sizeof (bgp));
-
-  if (rx_msg && p->started)
-    bmp_send_peer_up_notif_msg(p, bgp, pkt, pkt_size, rx_msg->data.buf,
-			       rx_msg->data.buf_size);
-}
-
-void
-bmp_put_recv_bgp_open_msg(const struct bgp_proto *bgp, const byte* pkt,
-  const size_t pkt_size)
-{
-  struct bmp_proto *p = g_bmp;
-
-  if (!p)
-  {
-    return;
-  }
-
-  struct bmp_peer_map_key key
-    = bmp_peer_map_key_create(bgp->remote_ip, bgp->remote_as);
-  const struct bmp_peer_map_entry *tx_msg
-    = bmp_peer_map_get(&p->peer_open_msg.tx_msg, key);
-
-  bmp_peer_map_insert(&p->peer_open_msg.rx_msg, key, pkt, pkt_size);
-
-  if (!tx_msg)
-    bmp_peer_map_insert(&p->bgp_peers, key, (const byte *) &bgp, sizeof (bgp));
-
-  if (tx_msg && p->started)
-    bmp_send_peer_up_notif_msg(p, bgp, tx_msg->data.buf, tx_msg->data.buf_size,
-			       pkt, pkt_size);
-}
-
-void
-bmp_route_monitor_update_in_pre_begin()
+static void
+bmp_route_monitor_put_update(struct bmp_proto *p, struct bmp_stream *bs, const byte *data, size_t length, btime timestamp)
 {
-  struct bmp_proto *p = g_bmp;
-
-  if (!p || !p->started)
-  {
-    return;
-  }
-
-  if (p->monitoring_rib.in_pre_policy == false)
-  {
-    return;
-  }
-
-  IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    !EMPTY_LIST(p->rt_table_in_pre_policy.update_msg_queue),
-    "Previous BMP route monitoring update not finished yet"
-  );
-
-  gettimeofday(&p->rt_table_in_pre_policy.update_begin_time,NULL);
-  init_list(&p->rt_table_in_pre_policy.update_msg_queue);
-  p->rt_table_in_pre_policy.update_msg_size = 0;
-  p->rt_table_in_pre_policy.update_in_progress = true;
+  struct bmp_data_node *upd_msg = mb_alloc(p->update_msg_mem_pool,
+                               sizeof (struct bmp_data_node));
+  upd_msg->data = mb_alloc(p->update_msg_mem_pool, length);
+  memcpy(upd_msg->data, data, length);
+  upd_msg->data_size = length;
+  add_tail(&p->update_msg_queue, &upd_msg->n);
+
+  /* Save some metadata */
+  struct bgp_proto *bgp = bs->bgp;
+  upd_msg->remote_as = bgp->remote_as;
+  upd_msg->remote_id = bgp->remote_id;
+  upd_msg->remote_ip = bgp->remote_ip;
+  upd_msg->timestamp = timestamp;
+  upd_msg->global_peer = bmp_is_peer_global_instance(bgp);
+  upd_msg->policy = bmp_stream_policy(bs);
+
+  /* Kick the commit */
+  if (!ev_active(p->update_ev))
+    ev_schedule(p->update_ev);
 }
 
-void
-bmp_route_monitor_put_update_in_pre_msg(const byte *data, const size_t data_size)
+static void
+bmp_route_monitor_notify(struct bmp_proto *p, struct bmp_stream *bs,
+			 const net_addr *n, const struct rte *new, const struct rte_src *src)
 {
-  struct bmp_proto *p = g_bmp;
-
-  if (!p || !p->started)
-  {
-    return;
-  }
-
-  if (p->monitoring_rib.in_pre_policy == false)
-  {
-    return;
-  }
+  byte buf[BGP_MAX_EXT_MSG_LENGTH];
+  byte *end = bgp_bmp_encode_rte(bs->sender, buf, n, new, src);
 
-  IF_COND_TRUE_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    !p->rt_table_in_pre_policy.update_in_progress,
-    "BMP route monitoring update not started yet"
-  );
+  btime delta_t = new ? current_time() - new->lastmod : 0;
+  btime timestamp = current_real_time() - delta_t;
 
-  struct bmp_data_node *upd_msg = mb_alloc(p->update_msg_mem_pool,
-                               sizeof (struct bmp_data_node));
-  upd_msg->data = mb_alloc(p->update_msg_mem_pool, data_size);
-  memcpy(upd_msg->data, data, data_size);
-  upd_msg->data_size = data_size;
-  p->rt_table_in_pre_policy.update_msg_size += data_size;
-  add_tail(&p->rt_table_in_pre_policy.update_msg_queue, &upd_msg->n);
+  if (end)
+    bmp_route_monitor_put_update(p, bs, buf, end - buf, timestamp);
+  else
+    log(L_WARN "%s: Cannot encode update for %N", p->p.name, n);
 }
 
-void
-bmp_route_monitor_update_in_pre_commit(const struct bgp_proto *bgp)
+static void
+bmp_route_monitor_commit(void *p_)
 {
-  struct bmp_proto *p = g_bmp;
-
-  if (!p || !p->started)
-  {
-    return;
-  }
+  struct bmp_proto *p = p_;
 
-  if (p->monitoring_rib.in_pre_policy == false)
-  {
+  if (!p->started)
     return;
-  }
 
-  const struct birdsock *sk = bmp_get_birdsock(bgp);
-  IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    sk,
-    "Failed to get bird socket from BGP proto"
-  );
-
-  const struct bgp_caps *remote_caps = bmp_get_bgp_remote_caps(bgp);
-  IF_PTR_IS_NULL_PRINT_ERR_MSG_AND_RETURN_OPT_VAL(
-    remote_caps,
-    "Failed to get remote capabilities from BGP proto"
-  );
-
-  bool is_global_instance_peer = bmp_is_peer_global_instance(bgp);
   buffer payload
-    = bmp_buffer_alloc(p->buffer_mpool,
-        p->rt_table_in_pre_policy.update_msg_size + DEFAULT_MEM_BLOCK_SIZE);
-
-  buffer update_msgs
-    = bmp_buffer_alloc(p->buffer_mpool,
-        p->rt_table_in_pre_policy.update_msg_size);
+    = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
 
-  struct bmp_data_node *data;
-  WALK_LIST(data, p->rt_table_in_pre_policy.update_msg_queue)
+  struct bmp_data_node *data, *data_next;
+  WALK_LIST_DELSAFE(data, data_next, p->update_msg_queue)
   {
-    bmp_put_data(&update_msgs, data->data, data->data_size);
     bmp_route_monitor_msg_serialize(&payload,
-      is_global_instance_peer, true /* TODO: Hardcoded pre-policy Adj-Rib-In */,
-      bgp->conn->received_as, bgp->remote_id, remote_caps->as4_support,
-      sk->daddr, bmp_buffer_data(&update_msgs), bmp_buffer_pos(&update_msgs),
-      p->rt_table_in_pre_policy.update_begin_time.tv_sec,
-      p->rt_table_in_pre_policy.update_begin_time.tv_usec);
+      data->global_peer, data->policy,
+      data->remote_as, data->remote_id, true,
+      data->remote_ip, data->data, data->data_size,
+      data->timestamp);
 
     bmp_schedule_tx_packet(p, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
 
     bmp_buffer_flush(&payload);
-    bmp_buffer_flush(&update_msgs);
-  }
-
-  bmp_buffer_free(&payload);
-  bmp_buffer_free(&update_msgs);
-}
 
-void
-bmp_route_monitor_update_in_pre_end()
-{
-  struct bmp_proto *p = g_bmp;
-
-  if (!p || !p->started)
-  {
-    return;
+    mb_free(data->data);
+    rem_node(&data->n);
+    mb_free(data);
   }
 
-  if (p->monitoring_rib.in_pre_policy == false)
-  {
-    return;
-  }
-
-  struct bmp_data_node *upd_msg;
-  struct bmp_data_node *upd_msg_next;
-  WALK_LIST_DELSAFE(upd_msg, upd_msg_next, p->rt_table_in_pre_policy.update_msg_queue)
-  {
-    mb_free(upd_msg->data);
-    rem_node((node *) upd_msg);
-    mb_free(upd_msg);
-  }
-
-  p->rt_table_in_pre_policy.update_in_progress = false;
+  bmp_buffer_free(&payload);
 }
 
 static void
-bmp_route_monitor_pre_policy_table_in_snapshot(struct channel *C)
+bmp_route_monitor_end_of_rib(struct bmp_proto *p, struct bmp_stream *bs)
 {
-  struct bmp_proto *p = g_bmp;
-
-  if (p->monitoring_rib.in_pre_policy == false)
-  {
-    return;
-  }
+  TRACE(D_PACKETS, "Sending END-OF-RIB for %s.%s", bs->bgp->p.name, bs->sender->c.name);
 
-  struct rtable *tab = C->in_table;
-  if (!tab)
-  {
-    return;
-  }
+  byte rx_end_payload[DEFAULT_MEM_BLOCK_SIZE];
+  byte *pos = bgp_create_end_mark_(bs->sender, rx_end_payload + BGP_HEADER_LENGTH);
+  memset(rx_end_payload + BGP_MSG_HDR_MARKER_POS, 0xff,
+	 BGP_MSG_HDR_MARKER_SIZE); // BGP UPDATE MSG marker
+  put_u16(rx_end_payload + BGP_MSG_HDR_LENGTH_POS, pos - rx_end_payload);
+  put_u8(rx_end_payload + BGP_MSG_HDR_TYPE_POS, PKT_UPDATE);
 
-  size_t cnt = 0;
-  struct proto *P;
-  struct fib_iterator fit;
-  memset(&fit, 0x00, sizeof (fit));
-  FIB_ITERATE_INIT(&fit, &tab->fib);
-  FIB_ITERATE_START(&tab->fib, &fit, net, n)
-  {
-    P = n->routes->sender->proto;
-    if (P->proto->class != PROTOCOL_BGP)
-    {
-      continue;
-    }
-
-    bmp_route_monitor_update_in_pre_begin();
-
-    rte *e;
-    for (e = n->routes; e; e = e->next)
-    {
-      bgp_rte_update_in_notify(C, n->n.addr, e, e->src);
-    }
-
-    bmp_route_monitor_update_in_pre_commit((struct bgp_proto *) P);
-    bmp_route_monitor_update_in_pre_end();
-    ++cnt;
-  }
-  FIB_ITERATE_END;
-
-  if (cnt > 0)
-  {
-    bmp_route_monitor_update_in_pre_begin();
-    byte rx_end_payload[DEFAULT_MEM_BLOCK_SIZE];
-    byte *pos
-      = bgp_create_end_mark((struct bgp_channel *) C, rx_end_payload
-                                                        + BGP_HEADER_LENGTH);
-    memset(rx_end_payload + BGP_MSG_HDR_MARKER_POS, 0xff,
-             BGP_MSG_HDR_MARKER_SIZE); // BGP UPDATE MSG marker
-    put_u16(rx_end_payload + BGP_MSG_HDR_LENGTH_POS, pos - rx_end_payload);
-    put_u8(rx_end_payload + BGP_MSG_HDR_TYPE_POS, PKT_UPDATE);
-    bmp_route_monitor_put_update_in_pre_msg(rx_end_payload, pos - rx_end_payload);
-    bmp_route_monitor_update_in_pre_commit((struct bgp_proto *) C->proto);
-    bmp_route_monitor_update_in_pre_end();
-  }
+  bmp_route_monitor_put_update(p, bs, rx_end_payload, pos - rx_end_payload, current_real_time());
 }
 
 static void
 bmp_send_peer_down_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
-  const byte* data, const size_t data_size)
+  const byte *data, const size_t data_size)
 {
   ASSERT(p->started);
 
@@ -853,69 +886,72 @@ bmp_send_peer_down_notif_msg(struct bmp_proto *p, const struct bgp_proto *bgp,
   bmp_buffer_free(&payload);
 }
 
-void
-bmp_peer_down(const struct bgp_proto *bgp, const int err_class, const byte *pkt,
-  size_t pkt_size)
+static void
+bmp_peer_down_(struct bmp_proto *p, const struct bgp_proto *bgp,
+	       int err_class, int err_code, int err_subcode, const byte *data, int length)
 {
-  struct bmp_proto *p = g_bmp;
-
-  if (!p)
-  {
+  if (!p->started)
     return;
-  }
 
-  struct bmp_peer_map_key key
-    = bmp_peer_map_key_create(bgp->remote_ip, bgp->remote_as);
+  struct bmp_peer *bp = bmp_find_peer(p, bgp);
+  if (!bp)
+    return;
 
-  bmp_peer_map_remove(&p->peer_open_msg.tx_msg, key);
-  bmp_peer_map_remove(&p->peer_open_msg.rx_msg, key);
-  bmp_peer_map_remove(&p->bgp_peers, key);
-  const size_t missing_bgp_hdr_size = BGP_MSG_HDR_MARKER_SIZE
-                                        + BGP_MSG_HDR_LENGTH_SIZE
-                                        + BGP_MSG_HDR_TYPE_SIZE;
+  TRACE(D_STATES, "Peer down for %s", bgp->p.name);
 
-  if (!p->started)
-    return;
+  uint bmp_code = 0;
+  uint fsm_code = 0;
 
-  buffer payload
-    = bmp_buffer_alloc(p->buffer_mpool, pkt_size + missing_bgp_hdr_size + 1);
-  if (pkt != NULL && pkt_size > 0)
+  switch (err_class)
   {
-    byte marker[BGP_MSG_HDR_MARKER_SIZE];
-    memset(marker, 0xff, BGP_MSG_HDR_MARKER_SIZE); // NOTIF MSG marker
-    if (!memcmp(pkt, marker, BGP_MSG_HDR_MARKER_SIZE))
-    {
-      // So it is received BGP PDU
-      bmp_put_u8(&payload, BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION);
-      bmp_put_data(&payload, pkt, pkt_size);
-    }
-    else
-    {
-      bmp_put_u8(&payload, BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION);
-      bmp_put_data(&payload, marker, BGP_MSG_HDR_MARKER_SIZE);
-      bmp_put_u16(&payload, pkt_size);
-      bmp_put_u8(&payload, PKT_NOTIFICATION);
-      bmp_put_data(&payload, pkt, pkt_size);
-    }
+  case BE_BGP_RX:
+    bmp_code = BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION;
+    break;
+
+  case BE_BGP_TX:
+  case BE_AUTO_DOWN:
+  case BE_MAN_DOWN:
+    bmp_code = BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION;
+    break;
+
+  default:
+    bmp_code = BMP_PEER_DOWN_REASON_REMOTE_NO_NOTIFICATION;
+    length = 0;
+    break;
   }
-  else
+
+  buffer payload = bmp_buffer_alloc(p->buffer_mpool, 1 + BGP_HEADER_LENGTH + 2 + length);
+  bmp_put_u8(&payload, bmp_code);
+
+  switch (bmp_code)
   {
-    // TODO: Handle De-configured Peer Down Reason Code
-    if (err_class == BE_SOCKET || err_class == BE_MISC)
-    {
-      bmp_put_u8(&payload, BMP_PEER_DOWN_REASON_REMOTE_NO_NOTIFICATION);
-    }
-    else
-    {
-      bmp_put_u8(&payload, BMP_PEER_DOWN_REASON_LOCAL_NO_NOTIFICATION);
-      // TODO: Fill in with appropriate FSM event code
-      bmp_put_u16(&payload, 0x00); // no relevant Event code is defined
-    }
+  case BMP_PEER_DOWN_REASON_LOCAL_BGP_NOTIFICATION:
+  case BMP_PEER_DOWN_REASON_REMOTE_BGP_NOTIFICATION:
+    bmp_put_bgp_hdr(&payload, BGP_HEADER_LENGTH + 2 + length, PKT_NOTIFICATION);
+    bmp_put_u8(&payload, err_code);
+    bmp_put_u8(&payload, err_subcode);
+    bmp_put_data(&payload, data, length);
+    break;
+
+  case BMP_PEER_DOWN_REASON_LOCAL_NO_NOTIFICATION:
+    bmp_put_u16(&payload, fsm_code);
+    break;
   }
 
   bmp_send_peer_down_notif_msg(p, bgp, bmp_buffer_data(&payload), bmp_buffer_pos(&payload));
 
   bmp_buffer_free(&payload);
+
+  bmp_remove_peer(p, bp);
+}
+
+void
+bmp_peer_down(const struct bgp_proto *bgp,
+	      int err_class, int code, int subcode, const byte *data, int length)
+{
+  struct bmp_proto *p; node *n;
+  WALK_LIST2(p, n, bmp_proto_list, bmp_node)
+    bmp_peer_down_(p, bgp, err_class, code, subcode, data, length);
 }
 
 static void
@@ -940,6 +976,73 @@ bmp_send_termination_msg(struct bmp_proto *p,
   bmp_buffer_free(&stream);
 }
 
+int
+bmp_preexport(struct channel *C UNUSED, rte *e)
+{
+  /* Reject non-direct routes */
+  if (e->src->proto != e->sender->proto)
+    return -1;
+
+  /* Reject non-BGP routes */
+  if (e->sender->channel != &channel_bgp)
+    return -1;
+
+  return 1;
+}
+
+static void
+bmp_rt_notify(struct proto *P, struct channel *c, struct network *net,
+		struct rte *new, struct rte *old)
+{
+  struct bmp_proto *p = (void *) P;
+
+  struct bgp_channel *src = (void *) (new ?: old)->sender;
+  struct bgp_proto *bgp = (void *) src->c.proto;
+  bool policy = (c->table == src->c.table);
+
+  /*
+   * We assume that we receive peer_up before the first route and peer_down
+   * synchronously with BGP session close. So if bmp_stream exists, the related
+   * BGP session is up and could be accessed. That may not be true in
+   * multithreaded setup.
+   */
+
+  struct bmp_stream *bs = bmp_find_stream(p, bgp, src->afi, policy);
+  if (!bs)
+    return;
+
+  bmp_route_monitor_notify(p, bs, net->n.addr, new, (new ?: old)->src);
+}
+
+static void
+bmp_feed_end(struct channel *c)
+{
+  struct bmp_proto *p = (void *) c->proto;
+
+  struct bmp_table *bt = bmp_find_table(p, c->table);
+  if (!bt)
+    return;
+
+  /*
+   * Unsynced streams are added in one moment during BMP session establishment,
+   * therefore we can assume that all unsynced streams (for given channel)
+   * already received full feed now and are synced.
+   *
+   * TODO: Use more efficent way to find bmp_stream from bmp_table
+   */
+
+  HASH_WALK(p->stream_map, next, bs)
+  {
+    if ((bs->table == bt) && !bs->sync)
+    {
+      bmp_route_monitor_end_of_rib(p, bs);
+      bs->sync = true;
+    }
+  }
+  HASH_WALK_END;
+}
+
+
 /**
  * bmp_startup - enter established state
  * @p: BMP instance
@@ -952,9 +1055,12 @@ bmp_startup(struct bmp_proto *p)
 {
   ASSERT(!p->started);
   p->started = true;
+  p->sock_err = 0;
 
   TRACE(D_EVENTS, "BMP session established");
 
+  proto_notify_state(&p->p, PS_UP);
+
   /* Send initiation message */
   buffer payload = bmp_buffer_alloc(p->buffer_mpool, DEFAULT_MEM_BLOCK_SIZE);
   bmp_init_msg_serialize(&payload, p->sys_descr, p->sys_name);
@@ -962,17 +1068,18 @@ bmp_startup(struct bmp_proto *p)
   bmp_buffer_free(&payload);
 
   /* Send Peer Up messages */
-  bmp_peer_map_walk(&p->peer_open_msg.tx_msg,
-		    bmp_peer_map_walk_tx_open_msg_and_send_peer_up_notif, p);
-
-  proto_notify_state(&p->p, PS_UP);
+  struct proto *peer;
+  WALK_LIST(peer, proto_list)
+    if ((peer->proto->class == PROTOCOL_BGP) && (peer->proto_state == PS_UP))
+      bmp_peer_init(p, (struct bgp_proto *) peer);
 }
 
 /**
  * bmp_down - leave established state
  * @p: BMP instance
  *
- * The bgp_down() function is called when the BMP session fails.
+ * The bgp_down() function is called when the BMP session fails. The caller is
+ * responsible for changing protocol state.
  */
 static void
 bmp_down(struct bmp_proto *p)
@@ -982,7 +1089,15 @@ bmp_down(struct bmp_proto *p)
 
   TRACE(D_EVENTS, "BMP session closed");
 
-  proto_notify_state(&p->p, PS_START);
+  /* Unregister existing peer structures */
+  HASH_WALK_DELSAFE(p->peer_map, next, bp)
+  {
+    bmp_remove_peer(p, bp);
+  }
+  HASH_WALK_END;
+
+  /* Removing peers should also remove all streams and tables */
+  ASSERT(!p->peer_map.count && !p->stream_map.count && !p->table_map.count);
 }
 
 /**
@@ -999,6 +1114,7 @@ bmp_connect(struct bmp_proto *p)
 
   sock *sk = sk_new(p->p.pool);
   sk->type = SK_TCP_ACTIVE;
+  sk->saddr = p->local_addr;
   sk->daddr = p->station_ip;
   sk->dport = p->station_port;
   sk->ttl = IP4_MAX_TTL;
@@ -1010,6 +1126,8 @@ bmp_connect(struct bmp_proto *p)
   p->sk = sk;
   sk->data = p;
 
+  TRACE(D_EVENTS, "Connecting to %I port %u", sk->daddr, sk->dport);
+
   int rc = sk_open(sk);
 
   if (rc < 0)
@@ -1018,12 +1136,14 @@ bmp_connect(struct bmp_proto *p)
   tm_start(p->connect_retry_timer, CONNECT_RETRY_TIME);
 }
 
-/* BMP connect successfull event - switch from Connect to Established state */
+/* BMP connect successful event - switch from Connect to Established state */
 static void
 bmp_connected(struct birdsock *sk)
 {
   struct bmp_proto *p = (void *) sk->data;
 
+  TRACE(D_EVENTS, "Connected");
+
   sk->rx_hook = bmp_rx;
   sk->tx_hook = bmp_tx;
   tm_stop(p->connect_retry_timer);
@@ -1037,6 +1157,8 @@ bmp_sock_err(sock *sk, int err)
 {
   struct bmp_proto *p = sk->data;
 
+  p->sock_err = err;
+
   if (err)
     TRACE(D_EVENTS, "Connection lost (%M)", err);
   else
@@ -1047,6 +1169,8 @@ bmp_sock_err(sock *sk, int err)
 
   bmp_close_socket(p);
   tm_start(p->connect_retry_timer, CONNECT_RETRY_TIME);
+
+  proto_notify_state(&p->p, PS_START);
 }
 
 /* BMP connect timeout event - switch from Idle/Connect state to Connect state */
@@ -1070,6 +1194,22 @@ bmp_close_socket(struct bmp_proto *p)
 }
 
 
+static void
+bmp_postconfig(struct proto_config *CF)
+{
+  struct bmp_config *cf = (void *) CF;
+
+  /* Do not check templates at all */
+  if (cf->c.class == SYM_TEMPLATE)
+    return;
+
+  if (ipa_zero(cf->station_ip))
+    cf_error("Station IP address not specified");
+
+  if (!cf->station_port)
+    cf_error("Station port number not specified");
+}
+
 /** Configuration handle section **/
 static struct proto *
 bmp_init(struct proto_config *CF)
@@ -1078,45 +1218,52 @@ bmp_init(struct proto_config *CF)
   struct bmp_proto *p = (void *) P;
   struct bmp_config *cf = (void *) CF;
 
+  P->rt_notify = bmp_rt_notify;
+  P->preexport = bmp_preexport;
+  P->feed_end = bmp_feed_end;
+
   p->cf = cf;
+  p->local_addr = cf->local_addr;
   p->station_ip = cf->station_ip;
   p->station_port = cf->station_port;
   strcpy(p->sys_descr, cf->sys_descr);
   strcpy(p->sys_name, cf->sys_name);
   p->monitoring_rib.in_pre_policy = cf->monitoring_rib_in_pre_policy;
   p->monitoring_rib.in_post_policy = cf->monitoring_rib_in_post_policy;
-  p->monitoring_rib.local = cf->monitoring_rib_local;
 
   return P;
 }
 
+/**
+ * bmp_start - initialize internal resources of BMP implementation.
+ * NOTE: It does not connect to BMP collector yet.
+ */
 static int
 bmp_start(struct proto *P)
 {
   struct bmp_proto *p = (void *) P;
 
-  log(L_DEBUG "Init BMP");
-
   p->buffer_mpool = rp_new(P->pool, "BMP Buffer");
   p->map_mem_pool = rp_new(P->pool, "BMP Map");
   p->tx_mem_pool = rp_new(P->pool, "BMP Tx");
   p->update_msg_mem_pool = rp_new(P->pool, "BMP Update");
-  p->tx_ev = ev_new_init(p->tx_mem_pool, bmp_fire_tx, p);
+  p->tx_ev = ev_new_init(p->p.pool, bmp_fire_tx, p);
+  p->update_ev = ev_new_init(p->p.pool, bmp_route_monitor_commit, p);
   p->connect_retry_timer = tm_new_init(p->p.pool, bmp_connection_retry, p, 0, 0);
   p->sk = NULL;
 
-  bmp_peer_map_init(&p->peer_open_msg.tx_msg, p->map_mem_pool);
-  bmp_peer_map_init(&p->peer_open_msg.rx_msg, p->map_mem_pool);
-  bmp_peer_map_init(&p->bgp_peers, p->map_mem_pool);
+  HASH_INIT(p->peer_map, P->pool, 4);
+  HASH_INIT(p->stream_map, P->pool, 4);
+  HASH_INIT(p->table_map, P->pool, 4);
 
   init_list(&p->tx_queue);
-  init_list(&p->rt_table_in_pre_policy.update_msg_queue);
+  init_list(&p->update_msg_queue);
   p->started = false;
+  p->sock_err = 0;
+  add_tail(&bmp_proto_list, &p->bmp_node);
 
   tm_start(p->connect_retry_timer, CONNECT_INIT_TIME);
 
-  g_bmp = p;
-
   return PS_START;
 }
 
@@ -1128,10 +1275,11 @@ bmp_shutdown(struct proto *P)
   if (p->started)
   {
     bmp_send_termination_msg(p, BMP_TERM_REASON_ADM);
-    p->started = false;
+    bmp_down(p);
   }
 
-  g_bmp = NULL;
+  p->sock_err = 0;
+  rem_node(&p->bmp_node);
 
   return PS_DOWN;
 }
@@ -1140,25 +1288,76 @@ static int
 bmp_reconfigure(struct proto *P, struct proto_config *CF)
 {
   struct bmp_proto *p = (void *) P;
-  const struct bmp_config *cf = (void *) CF;
+  const struct bmp_config *new = (void *) CF;
+  const struct bmp_config *old = p->cf;
 
-  log(L_WARN "Reconfiguring BMP is not supported");
+  int needs_restart = bstrcmp(new->sys_descr, old->sys_descr)
+    || bstrcmp(new->sys_name, old->sys_name)
+    || !ipa_equal(new->local_addr, old->local_addr)
+    || !ipa_equal(new->station_ip, old->station_ip)
+    || (new->station_port != old->station_port)
+    || (new->monitoring_rib_in_pre_policy != old->monitoring_rib_in_pre_policy)
+    || (new->monitoring_rib_in_post_policy != old->monitoring_rib_in_post_policy);
 
-  p->cf = cf;
+  /* If there is any change, restart the protocol */
+  if (needs_restart)
+    return 0;
+
+  /* We must update our copy of configuration ptr */
+  p->cf = new;
 
   return 1;
 }
 
+static void
+bmp_get_status(struct proto *P, byte *buf)
+{
+  struct bmp_proto *p = (void *) P;
+
+  if (P->proto_state == PS_DOWN)
+    bsprintf(buf, "Down");
+  else
+  {
+    const char *state = !p->started ? (!p->sk ? "Idle" : "Connect") : "Established";
+
+    if (!p->sock_err)
+      bsprintf(buf, "%s", state);
+    else
+      bsprintf(buf, "%-14s%s %M", state, "Error:", p->sock_err);
+  }
+}
+
+static void
+bmp_show_proto_info(struct proto *P)
+{
+  struct bmp_proto *p = (void *) P;
+
+  if (P->proto_state != PS_DOWN)
+  {
+    cli_msg(-1006, "  %-19s %I", "Station address:", p->station_ip);
+    cli_msg(-1006, "  %-19s %u", "Station port:", p->station_port);
+
+    if (!ipa_zero(p->local_addr))
+      cli_msg(-1006, "  %-19s %I", "Local address:", p->local_addr);
+
+    if (p->sock_err)
+      cli_msg(-1006, "  %-19s %M", "Last error:", p->sock_err);
+  }
+}
+
 struct protocol proto_bmp = {
   .name = "BMP",
   .template = "bmp%d",
   .class = PROTOCOL_BMP,
   .proto_size = sizeof(struct bmp_proto),
   .config_size = sizeof(struct bmp_config),
+  .postconfig = bmp_postconfig,
   .init = bmp_init,
   .start = bmp_start,
   .shutdown = bmp_shutdown,
   .reconfigure = bmp_reconfigure,
+  .get_status = bmp_get_status,
+  .show_proto_info = bmp_show_proto_info,
 };
 
 void