diff --git a/src/gateware/cxp.py b/src/gateware/cxp.py
index b828ce5..7447e4f 100644
--- a/src/gateware/cxp.py
+++ b/src/gateware/cxp.py
@@ -299,26 +299,25 @@ class CXP_Frame_Pipeline(Module, AutoCSR):
     # optimal stream packet size is 2 KiB - Section 9.5.2 (CXP-001-2021)
     def __init__(self, downconns, pmod_pads, packet_size=16384, n_buffer=1):
         n_downconn = len(downconns)
+        assert n_downconn > 0 and n_buffer > 0
+
+        # # #
         
         framebuffers = []
-        arr_csr = []
+        routing_ids = []
         cdr = ClockDomainsRenamer("cxp_gtx_rx")
         for i in range(n_buffer):
             # TODO: change this to rtio
             if i > 0:
                 name = "buffer_" + str(i) + "_routingid"
-                csr = CSRStorage(char_width, name=name, reset=i)
-                arr_csr.append(csr)
-                setattr(self, name, csr)
+                id = CSRStorage(char_width, name=name, reset=i)
+                routing_ids.append(id)
+                setattr(self, name, id)
 
             # roi_pipeline = cdr(ROI_Pipeline())
             # self.submodules += roi_pipeline
             # framebuffers.append(roi_pipeline.pipeline[0])
 
-            # DEBUG:
-            # self.comb += roi_pipeline.pipeline[-1].source.ack.eq(1)
-
-
             crc_checker = cdr(CXPCRC32_Checker())
 
             # TODO: handle full buffer gracefully
@@ -349,12 +348,35 @@ class CXP_Frame_Pipeline(Module, AutoCSR):
                 self.comb += pipeline[-1].source.ack.eq(1)
 
 
-        self.submodules.router = router = cdr(Frame_Packet_Router(downconns, framebuffers, packet_size, pmod_pads))
 
-        for i, csr in enumerate(arr_csr):
-            self.specials += MultiReg(csr.storage, router.routing_table[i], odomain="cxp_gtx_rx"),
+        #
+        #                        +---------+     +-------------+
+        # downconn pipline ----->|         |     |             |------> crc checker ------> raw stream data 
+        #                        | arbiter |---->| broadcaster |
+        # downconn pipline ----->|         |     |             |------> crc checker ------> raw stream data 
+        #                        +---------+     +-------------+
+        #
 
+
+        self.submodules.arbiter = arbiter = cdr(Stream_Arbiter(n_downconn))
+        self.submodules.broadcaster = broadcaster = cdr(Stream_Broadcaster(n_buffer))
+
+        # Connect pipeline
+        for i, d in enumerate(downconns):
+            # Assume downconns pipeline already marks the eop
+            self.comb += d.source.connect(arbiter.sinks[i])
+
+        self.comb += arbiter.source.connect(broadcaster.sink)
+
+        for i, fb in enumerate(framebuffers):
+            self.comb += broadcaster.sources[i].connect(fb.sink),
+
+        # Control interface
         # only the simple topology MASTER:ch0, extension:ch1,2,3 is supported right now
         active_extensions = Signal(max=n_downconn)
         self.sync += active_extensions.eq(reduce(add, [d.rx_ready.status for d in downconns[1:]]))
-        self.specials += MultiReg(active_extensions, router.n_ext_active, odomain="cxp_gtx_rx"),
+        self.specials += MultiReg(active_extensions, arbiter.n_ext_active, odomain="cxp_gtx_rx"),
+
+        for i, id in enumerate(routing_ids):
+            self.specials += MultiReg(id.storage, broadcaster.routing_ids[i], odomain="cxp_gtx_rx"),
+