diff --git a/adapters/common/openflow/utils.py b/adapters/common/openflow/utils.py
index b4c66cb..4547255 100644
--- a/adapters/common/openflow/utils.py
+++ b/adapters/common/openflow/utils.py
@@ -13,13 +13,269 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 #
-
+import structlog
 
 from adapters.protos import openflow_13_pb2 as ofp
+from hashlib import md5
 
+log = structlog.get_logger()
+
+# aliases
+ofb_field = ofp.ofp_oxm_ofb_field
+action = ofp.ofp_action
+
+# OFPAT_* shortcuts
 OUTPUT = ofp.OFPAT_OUTPUT
+COPY_TTL_OUT = ofp.OFPAT_COPY_TTL_OUT
+COPY_TTL_IN = ofp.OFPAT_COPY_TTL_IN
+SET_MPLS_TTL = ofp.OFPAT_SET_MPLS_TTL
+DEC_MPLS_TTL = ofp.OFPAT_DEC_MPLS_TTL
+PUSH_VLAN = ofp.OFPAT_PUSH_VLAN
+POP_VLAN = ofp.OFPAT_POP_VLAN
+PUSH_MPLS = ofp.OFPAT_PUSH_MPLS
+POP_MPLS = ofp.OFPAT_POP_MPLS
+SET_QUEUE = ofp.OFPAT_SET_QUEUE
+GROUP = ofp.OFPAT_GROUP
+SET_NW_TTL = ofp.OFPAT_SET_NW_TTL
+NW_TTL = ofp.OFPAT_DEC_NW_TTL
+SET_FIELD = ofp.OFPAT_SET_FIELD
+PUSH_PBB = ofp.OFPAT_PUSH_PBB
+POP_PBB = ofp.OFPAT_POP_PBB
+EXPERIMENTER = ofp.OFPAT_EXPERIMENTER
+
+# OFPXMT_OFB_* shortcuts (incomplete)
+IN_PORT = ofp.OFPXMT_OFB_IN_PORT
+IN_PHY_PORT = ofp.OFPXMT_OFB_IN_PHY_PORT
+METADATA = ofp.OFPXMT_OFB_METADATA
+ETH_DST = ofp.OFPXMT_OFB_ETH_DST
+ETH_SRC = ofp.OFPXMT_OFB_ETH_SRC
 ETH_TYPE = ofp.OFPXMT_OFB_ETH_TYPE
+VLAN_VID = ofp.OFPXMT_OFB_VLAN_VID
+VLAN_PCP = ofp.OFPXMT_OFB_VLAN_PCP
+IP_DSCP = ofp.OFPXMT_OFB_IP_DSCP
+IP_ECN = ofp.OFPXMT_OFB_IP_ECN
 IP_PROTO = ofp.OFPXMT_OFB_IP_PROTO
+IPV4_SRC = ofp.OFPXMT_OFB_IPV4_SRC
+IPV4_DST = ofp.OFPXMT_OFB_IPV4_DST
+TCP_SRC = ofp.OFPXMT_OFB_TCP_SRC
+TCP_DST = ofp.OFPXMT_OFB_TCP_DST
+UDP_SRC = ofp.OFPXMT_OFB_UDP_SRC
+UDP_DST = ofp.OFPXMT_OFB_UDP_DST
+SCTP_SRC = ofp.OFPXMT_OFB_SCTP_SRC
+SCTP_DST = ofp.OFPXMT_OFB_SCTP_DST
+ICMPV4_TYPE = ofp.OFPXMT_OFB_ICMPV4_TYPE
+ICMPV4_CODE = ofp.OFPXMT_OFB_ICMPV4_CODE
+ARP_OP = ofp.OFPXMT_OFB_ARP_OP
+ARP_SPA = ofp.OFPXMT_OFB_ARP_SPA
+ARP_TPA = ofp.OFPXMT_OFB_ARP_TPA
+ARP_SHA = ofp.OFPXMT_OFB_ARP_SHA
+ARP_THA = ofp.OFPXMT_OFB_ARP_THA
+IPV6_SRC = ofp.OFPXMT_OFB_IPV6_SRC
+IPV6_DST = ofp.OFPXMT_OFB_IPV6_DST
+IPV6_FLABEL = ofp.OFPXMT_OFB_IPV6_FLABEL
+ICMPV6_TYPE = ofp.OFPXMT_OFB_ICMPV6_TYPE
+ICMPV6_CODE = ofp.OFPXMT_OFB_ICMPV6_CODE
+IPV6_ND_TARGET = ofp.OFPXMT_OFB_IPV6_ND_TARGET
+OFB_IPV6_ND_SLL = ofp.OFPXMT_OFB_IPV6_ND_SLL
+IPV6_ND_TLL = ofp.OFPXMT_OFB_IPV6_ND_TLL
+MPLS_LABEL = ofp.OFPXMT_OFB_MPLS_LABEL
+MPLS_TC = ofp.OFPXMT_OFB_MPLS_TC
+MPLS_BOS = ofp.OFPXMT_OFB_MPLS_BOS
+PBB_ISID = ofp.OFPXMT_OFB_PBB_ISID
+TUNNEL_ID = ofp.OFPXMT_OFB_TUNNEL_ID
+IPV6_EXTHDR = ofp.OFPXMT_OFB_IPV6_EXTHDR
+
+# ofp_action_* shortcuts
+
+def output(port, max_len=ofp.OFPCML_MAX):
+    return action(
+        type=OUTPUT,
+        output=ofp.ofp_action_output(port=port, max_len=max_len)
+    )
+
+def mpls_ttl(ttl):
+    return action(
+        type=SET_MPLS_TTL,
+        mpls_ttl=ofp.ofp_action_mpls_ttl(mpls_ttl=ttl)
+    )
+
+def push_vlan(eth_type):
+    return action(
+        type=PUSH_VLAN,
+        push=ofp.ofp_action_push(ethertype=eth_type)
+    )
+
+def pop_vlan():
+    return action(
+        type=POP_VLAN
+    )
+
+def pop_mpls(eth_type):
+    return action(
+        type=POP_MPLS,
+        pop_mpls=ofp.ofp_action_pop_mpls(ethertype=eth_type)
+    )
+
+def group(group_id):
+    return action(
+        type=GROUP,
+        group=ofp.ofp_action_group(group_id=group_id)
+    )
+
+def nw_ttl(nw_ttl):
+    return action(
+        type=NW_TTL,
+        nw_ttl=ofp.ofp_action_nw_ttl(nw_ttl=nw_ttl)
+    )
+
+def set_field(field):
+    return action(
+        type=SET_FIELD,
+        set_field=ofp.ofp_action_set_field(
+            field=ofp.ofp_oxm_field(
+                oxm_class=ofp.OFPXMC_OPENFLOW_BASIC,
+                ofb_field=field))
+    )
+
+def experimenter(experimenter, data):
+    return action(
+        type=EXPERIMENTER,
+        experimenter=ofp.ofp_action_experimenter(
+            experimenter=experimenter, data=data)
+    )
+
+# ofb_field generators (incomplete set)
+
+def in_port(_in_port):
+    return ofb_field(type=IN_PORT, port=_in_port)
+
+def in_phy_port(_in_phy_port):
+    return ofb_field(type=IN_PHY_PORT, port=_in_phy_port)
+
+def metadata(_table_metadata):
+    return ofb_field(type=METADATA, table_metadata=_table_metadata)
+
+def eth_dst(_eth_dst):
+    return ofb_field(type=ETH_DST, table_metadata=_eth_dst)
+
+def eth_src(_eth_src):
+    return ofb_field(type=ETH_SRC, table_metadata=_eth_src)
+
+def eth_type(_eth_type):
+    return ofb_field(type=ETH_TYPE, eth_type=_eth_type)
+
+def vlan_vid(_vlan_vid):
+    return ofb_field(type=VLAN_VID, vlan_vid=_vlan_vid)
+
+def vlan_pcp(_vlan_pcp):
+    return ofb_field(type=VLAN_PCP, vlan_pcp=_vlan_pcp)
+
+def ip_dscp(_ip_dscp):
+    return ofb_field(type=IP_DSCP, ip_dscp=_ip_dscp)
+
+def ip_ecn(_ip_ecn):
+    return ofb_field(type=IP_ECN, ip_ecn=_ip_ecn)
+
+def ip_proto(_ip_proto):
+    return ofb_field(type=IP_PROTO, ip_proto=_ip_proto)
+
+def ipv4_src(_ipv4_src):
+    return ofb_field(type=IPV4_SRC, ipv4_src=_ipv4_src)
+
+def ipv4_dst(_ipv4_dst):
+    return ofb_field(type=IPV4_DST, ipv4_dst=_ipv4_dst)
+
+def tcp_src(_tcp_src):
+    return ofb_field(type=TCP_SRC, tcp_src=_tcp_src)
+
+def tcp_dst(_tcp_dst):
+    return ofb_field(type=TCP_DST, tcp_dst=_tcp_dst)
+
+def udp_src(_udp_src):
+    return ofb_field(type=UDP_SRC, udp_src=_udp_src)
+
+def udp_dst(_udp_dst):
+    return ofb_field(type=UDP_DST, udp_dst=_udp_dst)
+
+def sctp_src(_sctp_src):
+    return ofb_field(type=SCTP_SRC, sctp_src=_sctp_src)
+
+def sctp_dst(_sctp_dst):
+    return ofb_field(type=SCTP_DST, sctp_dst=_sctp_dst)
+
+def icmpv4_type(_icmpv4_type):
+    return ofb_field(type=ICMPV4_TYPE, icmpv4_type=_icmpv4_type)
+
+def icmpv4_code(_icmpv4_code):
+    return ofb_field(type=ICMPV4_CODE, icmpv4_code=_icmpv4_code)
+
+def arp_op(_arp_op):
+    return ofb_field(type=ARP_OP, arp_op=_arp_op)
+
+def arp_spa(_arp_spa):
+    return ofb_field(type=ARP_SPA, arp_spa=_arp_spa)
+
+def arp_tpa(_arp_tpa):
+    return ofb_field(type=ARP_TPA, arp_tpa=_arp_tpa)
+
+def arp_sha(_arp_sha):
+    return ofb_field(type=ARP_SHA, arp_sha=_arp_sha)
+
+def arp_tha(_arp_tha):
+    return ofb_field(type=ARP_THA, arp_tha=_arp_tha)
+
+def ipv6_src(_ipv6_src):
+    return ofb_field(type=IPV6_SRC, arp_tha=_ipv6_src)
+
+def ipv6_dst(_ipv6_dst):
+    return ofb_field(type=IPV6_DST, arp_tha=_ipv6_dst)
+
+def ipv6_flabel(_ipv6_flabel):
+    return ofb_field(type=IPV6_FLABEL, arp_tha=_ipv6_flabel)
+
+def ipmpv6_type(_icmpv6_type):
+    return ofb_field(type=ICMPV6_TYPE, arp_tha=_icmpv6_type)
+
+def icmpv6_code(_icmpv6_code):
+    return ofb_field(type=ICMPV6_CODE, arp_tha=_icmpv6_code)
+
+def ipv6_nd_target(_ipv6_nd_target):
+    return ofb_field(type=IPV6_ND_TARGET, arp_tha=_ipv6_nd_target)
+
+def ofb_ipv6_nd_sll(_ofb_ipv6_nd_sll):
+    return ofb_field(type=OFB_IPV6_ND_SLL, arp_tha=_ofb_ipv6_nd_sll)
+
+def ipv6_nd_tll(_ipv6_nd_tll):
+    return ofb_field(type=IPV6_ND_TLL, arp_tha=_ipv6_nd_tll)
+
+def mpls_label(_mpls_label):
+    return ofb_field(type=MPLS_LABEL, arp_tha=_mpls_label)
+
+def mpls_tc(_mpls_tc):
+    return ofb_field(type=MPLS_TC, arp_tha=_mpls_tc)
+
+def mpls_bos(_mpls_bos):
+    return ofb_field(type=MPLS_BOS, arp_tha=_mpls_bos)
+
+def pbb_isid(_pbb_isid):
+    return ofb_field(type=PBB_ISID, arp_tha=_pbb_isid)
+
+def tunnel_id(_tunnel_id):
+    return ofb_field(type=TUNNEL_ID, arp_tha=_tunnel_id)
+
+def ipv6_exthdr(_ipv6_exthdr):
+    return ofb_field(type=IPV6_EXTHDR, arp_tha=_ipv6_exthdr)
+
+
+# frequently used extractors:
+
+def get_actions(flow):
+    """Extract list of ofp_action objects from flow spec object"""
+    assert isinstance(flow, ofp.ofp_flow_stats)
+    # we have the following hard assumptions for now
+    for instruction in flow.instructions:
+        if instruction.type == ofp.OFPIT_APPLY_ACTIONS:
+            return instruction.actions.actions
 
 def get_ofb_fields(flow):
     assert isinstance(flow, ofp.ofp_flow_stats)
@@ -30,16 +286,213 @@
         ofb_fields.append(field.ofb_field)
     return ofb_fields
 
-def get_actions(flow):
-    """Extract list of ofp_action objects from flow spec object"""
-    assert isinstance(flow, ofp.ofp_flow_stats)
-    # we have the following hard assumptions for now
-    for instruction in flow.instructions:
-        if instruction.type == ofp.OFPIT_APPLY_ACTIONS:
-            return instruction.actions.actions
-
 def get_out_port(flow):
     for action in get_actions(flow):
         if action.type == OUTPUT:
             return action.output.port
     return None
+
+def get_in_port(flow):
+    for field in get_ofb_fields(flow):
+        if field.type == IN_PORT:
+            return field.port
+    return None
+
+def get_goto_table_id(flow):
+    for instruction in flow.instructions:
+        if instruction.type == ofp.OFPIT_GOTO_TABLE:
+            return instruction.goto_table.table_id
+    return None
+
+def get_metadata(flow):
+    ''' legacy get method (only want lower 32 bits '''
+    for field in get_ofb_fields(flow):
+        if field.type == METADATA:
+            return field.table_metadata & 0xffffffff
+    return None
+
+def get_metadata_64_bit(flow):
+    for field in get_ofb_fields(flow):
+        if field.type == METADATA:
+            return field.table_metadata
+    return None
+
+
+def get_port_number_from_metadata(flow):
+    """
+    The port number (UNI on ONU) is in the lower 32-bits of metadata and
+    the inner_tag is in the upper 32-bits
+
+    This is set in the ONOS OltPipeline as a metadata field
+    """
+    md = get_metadata_64_bit(flow)
+
+    if md is None:
+        return None
+
+    if md <= 0xffffffff:
+        log.warn('onos-upgrade-suggested',
+                 netadata=md,
+                 message='Legacy MetaData detected form OltPipeline')
+        return md
+
+    return md & 0xffffffff
+
+
+def get_inner_tag_from_metadata(flow):
+    """
+    The port number (UNI on ONU) is in the lower 32-bits of metadata and
+    the inner_tag is in the upper 32-bits
+
+    This is set in the ONOS OltPipeline as a metadata field
+    """
+    md = get_metadata_64_bit(flow)
+
+    if md is None:
+        return None
+
+    if md <= 0xffffffff:
+        log.warn('onos-upgrade-suggested',
+                 netadata=md,
+                 message='Legacy MetaData detected form OltPipeline')
+        return md
+
+    return (md >> 32) & 0xffffffff
+
+
+# test and extract next table and group information
+def has_next_table(flow):
+    return get_goto_table_id(flow) is not None
+
+def get_group(flow):
+    for action in get_actions(flow):
+        if action.type == GROUP:
+            return action.group.group_id
+    return None
+
+def has_group(flow):
+    return get_group(flow) is not None
+
+def mk_oxm_fields(match_fields):
+    oxm_fields=[
+        ofp.ofp_oxm_field(
+            oxm_class=ofp.OFPXMC_OPENFLOW_BASIC,
+            ofb_field=field
+        ) for field in match_fields
+    ]
+
+    return oxm_fields
+
+def mk_instructions_from_actions(actions):
+    instructions_action = ofp.ofp_instruction_actions()
+    instructions_action.actions.extend(actions)
+    instruction = ofp.ofp_instruction(type=ofp.OFPIT_APPLY_ACTIONS,
+                                      actions=instructions_action)
+    return [instruction]
+
+def mk_simple_flow_mod(match_fields, actions, command=ofp.OFPFC_ADD,
+                       next_table_id=None, **kw):
+    """
+    Convenience function to generare ofp_flow_mod message with OXM BASIC match
+    composed from the match_fields, and single APPLY_ACTIONS instruction with
+    a list if ofp_action objects.
+    :param match_fields: list(ofp_oxm_ofb_field)
+    :param actions: list(ofp_action)
+    :param command: one of OFPFC_*
+    :param kw: additional keyword-based params to ofp_flow_mod
+    :return: initialized ofp_flow_mod object
+    """
+    instructions = [
+        ofp.ofp_instruction(
+            type=ofp.OFPIT_APPLY_ACTIONS,
+            actions=ofp.ofp_instruction_actions(actions=actions)
+        )
+    ]
+    if next_table_id is not None:
+        instructions.append(ofp.ofp_instruction(
+            type=ofp.OFPIT_GOTO_TABLE,
+            goto_table=ofp.ofp_instruction_goto_table(table_id=next_table_id)
+        ))
+
+    return ofp.ofp_flow_mod(
+        command=command,
+        match=ofp.ofp_match(
+            type=ofp.OFPMT_OXM,
+            oxm_fields=[
+                ofp.ofp_oxm_field(
+                    oxm_class=ofp.OFPXMC_OPENFLOW_BASIC,
+                    ofb_field=field
+                ) for field in match_fields
+            ]
+        ),
+        instructions=instructions,
+        **kw
+    )
+
+
+def mk_multicast_group_mod(group_id, buckets, command=ofp.OFPGC_ADD):
+    group = ofp.ofp_group_mod(
+        command=command,
+        type=ofp.OFPGT_ALL,
+        group_id=group_id,
+        buckets=buckets
+    )
+    return group
+
+
+def hash_flow_stats(flow):
+    """
+    Return unique 64-bit integer hash for flow covering the following
+    attributes: 'table_id', 'priority', 'flags', 'cookie', 'match', '_instruction_string'
+    """
+    _instruction_string = ""
+    for _instruction in flow.instructions:
+        _instruction_string += _instruction.SerializeToString()
+
+    hex = md5('{},{},{},{},{},{}'.format(
+        flow.table_id,
+        flow.priority,
+        flow.flags,
+        flow.cookie,
+        flow.match.SerializeToString(),
+        _instruction_string
+    )).hexdigest()
+    return int(hex[:16], 16)
+
+
+def flow_stats_entry_from_flow_mod_message(mod):
+    flow = ofp.ofp_flow_stats(
+        table_id=mod.table_id,
+        priority=mod.priority,
+        idle_timeout=mod.idle_timeout,
+        hard_timeout=mod.hard_timeout,
+        flags=mod.flags,
+        cookie=mod.cookie,
+        match=mod.match,
+        instructions=mod.instructions
+    )
+    flow.id = hash_flow_stats(flow)
+    return flow
+
+
+def group_entry_from_group_mod(mod):
+    group = ofp.ofp_group_entry(
+        desc=ofp.ofp_group_desc(
+            type=mod.type,
+            group_id=mod.group_id,
+            buckets=mod.buckets
+        ),
+        stats=ofp.ofp_group_stats(
+            group_id=mod.group_id
+            # TODO do we need to instantiate bucket bins?
+        )
+    )
+    return group
+
+
+def mk_flow_stat(**kw):
+    return flow_stats_entry_from_flow_mod_message(mk_simple_flow_mod(**kw))
+
+
+def mk_group_stat(**kw):
+    return group_entry_from_group_mod(mk_multicast_group_mod(**kw))
\ No newline at end of file
diff --git a/adapters/iadapter.py b/adapters/iadapter.py
index 8cdcd6e..0d32096 100644
--- a/adapters/iadapter.py
+++ b/adapters/iadapter.py
@@ -159,8 +159,9 @@
         log.info('bulk-flow-update', device_id=device.id,
                  flows=flows, groups=groups)
         assert len(groups.items) == 0
-        handler = self.devices_handlers[device.id]
-        return handler.update_flow_table(flows.items)
+        reactor.callLater(0, self.devices_handlers[device.id].update_flow_table, flows.items)
+        return device
+
 
     def update_flows_incrementally(self, device, flow_changes, group_changes):
         log.info('incremental-flow-update', device_id=device.id,
@@ -172,11 +173,12 @@
         handler = self.devices_handlers[device.id]
         # Remove flows
         if len(flow_changes.to_remove.items) != 0:
-            handler.remove_from_flow_table(flow_changes.to_remove.items)
+            reactor.callLater(0, handler.remove_from_flow_table, flow_changes.to_remove.items)
 
         # Add flows
         if len(flow_changes.to_add.items) != 0:
-            handler.add_to_flow_table(flow_changes.to_add.items)
+            reactor.callLater(0, handler.add_to_flow_table, flow_changes.to_add.items)
+        return device
 
     def update_pm_config(self, device, pm_config):
         log.info("adapter-update-pm-config", device=device,
diff --git a/adapters/kafka/adapter_request_facade.py b/adapters/kafka/adapter_request_facade.py
index f4898a3..3009206 100644
--- a/adapters/kafka/adapter_request_facade.py
+++ b/adapters/kafka/adapter_request_facade.py
@@ -42,6 +42,7 @@
 from adapters.common.utils.registry import registry
 from adapters.common.utils.id_generation import create_cluster_device_id
 from adapters.protos.core_adapter_pb2 import IntType
+from adapters.protos.openflow_13_pb2 import FlowChanges, FlowGroups, Flows, FlowGroupChanges
 import re
 
 
@@ -167,10 +168,38 @@
         return self.adapter.get_device_details(device)
 
     def update_flows_bulk(self, device, flows, groups):
-        return self.adapter.update_flows_bulk(device, flows, groups)
+        d = Device()
+        if device:
+            device.Unpack(d)
+        else:
+            return (False, d)
+
+        f = Flows()
+        if flows:
+            flows.Unpack(f)
+
+        g = FlowGroups()
+        if groups:
+            groups.Unpack(g)
+
+        return (True, self.adapter.update_flows_bulk(d, f, g))
 
     def update_flows_incrementally(self, device, flow_changes, group_changes):
-        return self.adapter.update_flows_incrementally(device, flow_changes, group_changes)
+        d = Device()
+        if device:
+            device.Unpack(d)
+        else:
+            return (False, d)
+
+        f = FlowChanges()
+        if flow_changes:
+            flow_changes.Unpack(f)
+
+        g = FlowGroupChanges()
+        if group_changes:
+            group_changes.Unpack(g)
+
+        return (True, self.adapter.update_flows_incrementally(d, f, g))
 
     def suppress_alarm(self, filter):
         return self.adapter.suppress_alarm(filter)
diff --git a/adapters/ponsim_olt/ponsim_olt.py b/adapters/ponsim_olt/ponsim_olt.py
index e47d14d..1806a33 100644
--- a/adapters/ponsim_olt/ponsim_olt.py
+++ b/adapters/ponsim_olt/ponsim_olt.py
@@ -358,12 +358,12 @@
         cap = OFPPF_1GB_FD | OFPPF_FIBER
         return PortCapability(
             port = LogicalPort (
-                id='nni',
+                # id='nni',
                 ofp_port=ofp_port(
-                    port_no=port_no,
+                    # port_no=port_no,
                     hw_addr=mac_str_to_tuple(
-                    '00:00:00:00:00:%02x' % self.ofp_port_no),
-                    name='nni',
+                    '00:00:00:00:00:%02x' % port_no),
+                    # name='nni',
                     config=0,
                     state=OFPPS_LIVE,
                     curr=cap,
@@ -371,8 +371,10 @@
                     peer=cap,
                     curr_speed=OFPPF_1GB_FD,
                     max_speed=OFPPF_1GB_FD
-                )
-            )
+                ),
+                device_id=device.id,
+                device_port_no=port_no
+        )
         )
 
     # TODO - change for core 2.0
@@ -458,34 +460,12 @@
         self.log.info('stopped-receiving-grpc-frames')
 
 
-    # VOLTHA's flow decomposition removes the information about which flows
-    # are trap flows where traffic should be forwarded to the controller.
-    # We'll go through the flows and change the output port of flows that we
-    # know to be trap flows to the OF CONTROLLER port.
+    @inlineCallbacks
     def update_flow_table(self, flows):
-        stub = ponsim_pb2.PonSimStub(self.get_channel())
-        self.log.info('pushing-olt-flow-table')
-        for flow in flows:
-            classifier_info = {}
-            for field in fd.get_ofb_fields(flow):
-                if field.type == fd.ETH_TYPE:
-                    classifier_info['eth_type'] = field.eth_type
-                    self.log.debug('field-type-eth-type',
-                                   eth_type=classifier_info['eth_type'])
-                elif field.type == fd.IP_PROTO:
-                    classifier_info['ip_proto'] = field.ip_proto
-                    self.log.debug('field-type-ip-proto',
-                                   ip_proto=classifier_info['ip_proto'])
-            if ('ip_proto' in classifier_info and (
-                    classifier_info['ip_proto'] == 17 or
-                    classifier_info['ip_proto'] == 2)) or (
-                    'eth_type' in classifier_info and
-                    classifier_info['eth_type'] == 0x888e):
-                for action in fd.get_actions(flow):
-                    if action.type == ofp.OFPAT_OUTPUT:
-                        action.output.port = ofp.OFPP_CONTROLLER
-            self.log.info('out_port', out_port=fd.get_out_port(flow))
+        yield self.get_channel()
+        stub = ponsim_pb2.PonSimStub(self.channel)
 
+        self.log.info('pushing-olt-flow-table')
         stub.UpdateFlowTable(FlowTable(
             port=0,
             flows=flows
diff --git a/adapters/ponsim_onu/ponsim_onu.py b/adapters/ponsim_onu/ponsim_onu.py
index a9d3710..19775cb 100644
--- a/adapters/ponsim_onu/ponsim_onu.py
+++ b/adapters/ponsim_onu/ponsim_onu.py
@@ -114,20 +114,21 @@
         yield self.adapter_agent.device_state_update(device.id, connect_status=ConnectStatus.REACHABLE, oper_status=OperStatus.ACTIVE)
 
 
+    # TODO: Return only port specific info
     def get_ofp_port_info(self, device, port_no):
         # Since the adapter created the device port then it has the reference of the port to
-        # return the capability.   TODO:  Do a lookup on the NNI port number and return the
+        # return the capability.   TODO:  Do a lookup on the UNI port number and return the
         # appropriate attributes
         self.log.info('get_ofp_port_info', port_no=port_no, device_id=device.id)
         # port_no = device.proxy_address.channel_id
         cap = OFPPF_1GB_FD | OFPPF_FIBER
         return PortCapability(
             port = LogicalPort (
-                id='uni-{}'.format(port_no),
+                # id='uni-{}'.format(port_no),
                 ofp_port=ofp_port(
-                    port_no=port_no,
+                    # port_no=port_no,
                     hw_addr=mac_str_to_tuple('00:00:00:00:00:%02x' % port_no),
-                    name='uni-{}'.format(port_no),
+                    # name='uni-{}'.format(port_no),
                     config=0,
                     state=OFPPS_LIVE,
                     curr=cap,
@@ -135,7 +136,9 @@
                     peer=cap,
                     curr_speed=OFPPF_1GB_FD,
                     max_speed=OFPPF_1GB_FD
-                )
+                ),
+                device_id=device.id,
+                device_port_no=port_no
             )
         )
 
diff --git a/protos/openflow_13.proto b/protos/openflow_13.proto
index 47d881b..e242e76 100644
--- a/protos/openflow_13.proto
+++ b/protos/openflow_13.proto
@@ -2278,6 +2278,7 @@
 message FlowGroupChanges {
     FlowGroups to_add = 1;
     FlowGroups to_remove = 2;
+    FlowGroups to_update = 3;
 }
 
 message PacketIn {
diff --git a/rw_core/core/adapter_proxy.go b/rw_core/core/adapter_proxy.go
index 82c6dec..643c9de 100644
--- a/rw_core/core/adapter_proxy.go
+++ b/rw_core/core/adapter_proxy.go
@@ -22,6 +22,7 @@
 	"github.com/opencord/voltha-go/common/log"
 	"github.com/opencord/voltha-go/kafka"
 	ca "github.com/opencord/voltha-go/protos/core_adapter"
+	"github.com/opencord/voltha-go/protos/openflow_13"
 	"github.com/opencord/voltha-go/protos/voltha"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -249,14 +250,50 @@
 	return nil, nil
 }
 
-func (ap *AdapterProxy) UpdateFlowsBulk(device voltha.Device, flows voltha.Flows, groups voltha.FlowGroups) error {
-	log.Debug("UpdateFlowsBulk")
-	return nil
+func (ap *AdapterProxy) UpdateFlowsBulk(device *voltha.Device, flows *voltha.Flows, groups *voltha.FlowGroups) error {
+	log.Debugw("UpdateFlowsBulk", log.Fields{"deviceId": device.Id})
+	topic := kafka.Topic{Name: device.Type}
+	rpc := "update_flows_bulk"
+	args := make([]*kafka.KVArg, 3)
+	args[0] = &kafka.KVArg{
+		Key:   "device",
+		Value: device,
+	}
+	args[1] = &kafka.KVArg{
+		Key:   "flows",
+		Value: flows,
+	}
+	args[2] = &kafka.KVArg{
+		Key:   "groups",
+		Value: groups,
+	}
+
+	success, result := ap.kafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+	log.Debugw("UpdateFlowsBulk-response", log.Fields{"deviceid": device.Id, "success": success})
+	return unPackResponse(rpc, device.Id, success, result)
 }
 
-func (ap *AdapterProxy) UpdateFlowsIncremental(device voltha.Device, flowChanges voltha.Flows, groupChanges voltha.FlowGroups) error {
-	log.Debug("UpdateFlowsIncremental")
-	return nil
+func (ap *AdapterProxy) UpdateFlowsIncremental(device *voltha.Device, flowChanges *openflow_13.FlowChanges, groupChanges *openflow_13.FlowGroupChanges) error {
+	log.Debugw("UpdateFlowsIncremental", log.Fields{"deviceId": device.Id})
+	topic := kafka.Topic{Name: device.Type}
+	rpc := "update_flows_bulk"
+	args := make([]*kafka.KVArg, 3)
+	args[0] = &kafka.KVArg{
+		Key:   "device",
+		Value: device,
+	}
+	args[1] = &kafka.KVArg{
+		Key:   "flow_changes",
+		Value: flowChanges,
+	}
+	args[2] = &kafka.KVArg{
+		Key:   "group_changes",
+		Value: groupChanges,
+	}
+
+	success, result := ap.kafkaProxy.InvokeRPC(nil, rpc, &topic, true, args...)
+	log.Debugw("UpdateFlowsIncremental-response", log.Fields{"deviceid": device.Id, "success": success})
+	return unPackResponse(rpc, device.Id, success, result)
 }
 
 func (ap *AdapterProxy) UpdatePmConfig(device voltha.Device, pmConfigs voltha.PmConfigs) error {
diff --git a/rw_core/core/adapter_request_handler.go b/rw_core/core/adapter_request_handler.go
index 0c0609e..e7f69f5 100644
--- a/rw_core/core/adapter_request_handler.go
+++ b/rw_core/core/adapter_request_handler.go
@@ -17,6 +17,7 @@
 
 import (
 	"errors"
+	"github.com/gogo/protobuf/proto"
 	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/empty"
 	"github.com/opencord/voltha-go/common/log"
@@ -25,7 +26,6 @@
 	"github.com/opencord/voltha-go/protos/voltha"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
-	"reflect"
 )
 
 type AdapterRequestHandlerProxy struct {
@@ -83,7 +83,7 @@
 	}
 
 	// Get the device via the device manager
-	if device, err := rhp.deviceMgr.getDevice(pID.Id); err != nil {
+	if device, err := rhp.deviceMgr.GetDevice(pID.Id); err != nil {
 		return nil, status.Errorf(codes.NotFound, "%s", err.Error())
 	} else {
 		return device, nil
@@ -96,16 +96,17 @@
 	//		First retrieve the most up to date device info
 	var currentDevice *voltha.Device
 	var err error
-	if currentDevice, err = rhp.deviceMgr.getDevice(device.Id); err != nil {
+	if currentDevice, err = rhp.deviceMgr.GetDevice(device.Id); err != nil {
 		return nil, err
 	}
-	cloned := reflect.ValueOf(currentDevice).Elem().Interface().(voltha.Device)
+	cloned := proto.Clone(currentDevice).(*voltha.Device)
 	cloned.Root = device.Root
 	cloned.Vendor = device.Vendor
 	cloned.Model = device.Model
 	cloned.SerialNumber = device.SerialNumber
 	cloned.MacAddress = device.MacAddress
-	return &cloned, nil
+	cloned.Vlan = device.Vlan
+	return cloned, nil
 }
 
 func (rhp *AdapterRequestHandlerProxy) DeviceUpdate(args []*ca.Argument) (*empty.Empty, error) {
diff --git a/rw_core/core/device_agent.go b/rw_core/core/device_agent.go
index 52ab584..e045fc9 100644
--- a/rw_core/core/device_agent.go
+++ b/rw_core/core/device_agent.go
@@ -17,16 +17,18 @@
 
 import (
 	"context"
-	"reflect"
-	"sync"
-
+	"fmt"
 	"github.com/gogo/protobuf/proto"
 	"github.com/opencord/voltha-go/common/log"
 	"github.com/opencord/voltha-go/db/model"
 	"github.com/opencord/voltha-go/protos/core_adapter"
+	ofp "github.com/opencord/voltha-go/protos/openflow_13"
 	"github.com/opencord/voltha-go/protos/voltha"
+	fu "github.com/opencord/voltha-go/rw_core/utils"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"reflect"
+	"sync"
 )
 
 type DeviceAgent struct {
@@ -37,6 +39,8 @@
 	clusterDataProxy *model.Proxy
 	deviceProxy      *model.Proxy
 	exitChannel      chan int
+	flowProxy        *model.Proxy
+	groupProxy       *model.Proxy
 	lockDevice       sync.RWMutex
 }
 
@@ -48,6 +52,13 @@
 	cloned := (proto.Clone(device)).(*voltha.Device)
 	cloned.Id = CreateDeviceId()
 	cloned.AdminState = voltha.AdminState_PREPROVISIONED
+	cloned.FlowGroups = &ofp.FlowGroups{Items: nil}
+	cloned.Flows = &ofp.Flows{Items: nil}
+	if !device.GetRoot() && device.ProxyAddress != nil {
+		// Set the default vlan ID to the one specified by the parent adapter.  It can be
+		// overwritten by the child adapter during a device update request
+		cloned.Vlan = device.ProxyAddress.ChannelId
+	}
 	agent.deviceId = cloned.Id
 	agent.lastData = cloned
 	agent.deviceMgr = deviceMgr
@@ -68,6 +79,17 @@
 	}
 	agent.deviceProxy = agent.clusterDataProxy.Root.GetProxy("/devices/"+agent.deviceId, false)
 	agent.deviceProxy.RegisterCallback(model.POST_UPDATE, agent.processUpdate, nil)
+
+	agent.flowProxy = agent.clusterDataProxy.Root.GetProxy(
+		fmt.Sprintf("/devices/%s/flows", agent.deviceId),
+		false)
+	agent.groupProxy = agent.clusterDataProxy.Root.GetProxy(
+		fmt.Sprintf("/devices/%s/flow_groups", agent.deviceId),
+		false)
+
+	agent.flowProxy.RegisterCallback(model.POST_UPDATE, agent.flowTableUpdated)
+	//agent.groupProxy.RegisterCallback(model.POST_UPDATE, agent.groupTableUpdated)
+
 	log.Debug("device-agent-started")
 }
 
@@ -80,7 +102,7 @@
 	log.Debug("device-agent-stopped")
 }
 
-// getDevice retrieves the latest device information from the data model
+// GetDevice retrieves the latest device information from the data model
 func (agent *DeviceAgent) getDevice() (*voltha.Device, error) {
 	agent.lockDevice.Lock()
 	defer agent.lockDevice.Unlock()
@@ -146,6 +168,51 @@
 	return nil
 }
 
+func (agent *DeviceAgent) updateFlows(flows []*ofp.OfpFlowStats) error {
+	agent.lockDevice.Lock()
+	defer agent.lockDevice.Unlock()
+	log.Debugw("updateFlows", log.Fields{"deviceId": agent.deviceId, "flows": flows})
+	var oldData *voltha.Flows
+	if storedData, err := agent.getDeviceWithoutLock(); err != nil {
+		return status.Errorf(codes.NotFound, "%s", agent.deviceId)
+	} else {
+		oldData = proto.Clone(storedData.Flows).(*voltha.Flows)
+		log.Debugw("updateFlows", log.Fields{"deviceId": agent.deviceId, "flows": flows, "old": oldData})
+		// store the changed data
+		storedData.Flows.Items = flows
+		afterUpdate := agent.clusterDataProxy.Update("/devices/"+agent.deviceId, storedData, false, "")
+		if afterUpdate == nil {
+			return status.Errorf(codes.Internal, "%s", agent.deviceId)
+		}
+
+		// For now, force the callback to occur
+		go agent.flowTableUpdated(oldData, &ofp.Flows{Items: flows})
+		return nil
+	}
+}
+
+func (agent *DeviceAgent) updateGroups(groups []*ofp.OfpGroupEntry) error {
+	agent.lockDevice.Lock()
+	defer agent.lockDevice.Unlock()
+	var oldData *voltha.FlowGroups
+	log.Debugw("updateGroups", log.Fields{"deviceId": agent.deviceId, "groups": groups})
+	if storedData, err := agent.getDeviceWithoutLock(); err != nil {
+		return status.Errorf(codes.NotFound, "%s", agent.deviceId)
+	} else {
+		oldData = proto.Clone(storedData.FlowGroups).(*voltha.FlowGroups)
+		// store the changed data
+		storedData.FlowGroups.Items = groups
+		afterUpdate := agent.clusterDataProxy.Update("/devices/"+agent.deviceId, storedData, false, "")
+		if afterUpdate == nil {
+			return status.Errorf(codes.Internal, "%s", agent.deviceId)
+		}
+
+		// For now, force the callback to occur
+		go agent.groupTableUpdated(oldData, &ofp.FlowGroups{Items: groups})
+		return nil
+	}
+}
+
 //disableDevice disable a device
 func (agent *DeviceAgent) disableDevice(ctx context.Context) error {
 	agent.lockDevice.Lock()
@@ -257,7 +324,7 @@
 func (agent *DeviceAgent) getPorts(ctx context.Context, portType voltha.Port_PortType) *voltha.Ports {
 	log.Debugw("getPorts", log.Fields{"id": agent.deviceId, "portType": portType})
 	ports := &voltha.Ports{}
-	if device, _ := agent.deviceMgr.getDevice(agent.deviceId); device != nil {
+	if device, _ := agent.deviceMgr.GetDevice(agent.deviceId); device != nil {
 		for _, port := range device.Ports {
 			if port.Type == portType {
 				ports.Items = append(ports.Items, port)
@@ -271,7 +338,7 @@
 // parent device
 func (agent *DeviceAgent) getSwitchCapability(ctx context.Context) (*core_adapter.SwitchCapability, error) {
 	log.Debugw("getSwitchCapability", log.Fields{"deviceId": agent.deviceId})
-	if device, err := agent.deviceMgr.getDevice(agent.deviceId); device == nil {
+	if device, err := agent.deviceMgr.GetDevice(agent.deviceId); device == nil {
 		return nil, err
 	} else {
 		var switchCap *core_adapter.SwitchCapability
@@ -288,7 +355,7 @@
 // device
 func (agent *DeviceAgent) getPortCapability(ctx context.Context, portNo uint32) (*core_adapter.PortCapability, error) {
 	log.Debugw("getPortCapability", log.Fields{"deviceId": agent.deviceId})
-	if device, err := agent.deviceMgr.getDevice(agent.deviceId); device == nil {
+	if device, err := agent.deviceMgr.GetDevice(agent.deviceId); device == nil {
 		return nil, err
 	} else {
 		var portCap *core_adapter.PortCapability
@@ -500,6 +567,164 @@
 	}
 }
 
+//flowTableUpdated is the callback after flows have been updated in the model to push them
+//to the adapters
+func (agent *DeviceAgent) flowTableUpdated(args ...interface{}) interface{} {
+	log.Debugw("flowTableUpdated-callback", log.Fields{"argsLen": len(args)})
+
+	agent.lockDevice.Lock()
+	defer agent.lockDevice.Unlock()
+
+	var previousData *voltha.Flows
+	var latestData *voltha.Flows
+
+	var ok bool
+	if previousData, ok = args[0].(*ofp.Flows); !ok {
+		log.Errorw("invalid-args", log.Fields{"args0": args[0]})
+		return nil
+	}
+	if latestData, ok = args[1].(*ofp.Flows); !ok {
+		log.Errorw("invalid-args", log.Fields{"args1": args[1]})
+		return nil
+	}
+
+	// Sanity check - should not happen as this is already handled in logical device agent
+	if reflect.DeepEqual(previousData.Items, latestData.Items) {
+		log.Debugw("flow-update-not-required", log.Fields{"previous": previousData.Items, "new": latestData.Items})
+		return nil
+	}
+
+	var device *voltha.Device
+	var err error
+	if device, err = agent.getDeviceWithoutLock(); err != nil {
+		log.Errorw("no-device", log.Fields{"id": agent.deviceId, "error": err})
+		return nil
+	}
+	groups := device.FlowGroups
+
+	// Send update to adapters
+	// Check whether the device supports incremental flow changes
+	// Assume false for test
+	acceptsAddRemoveFlowUpdates := false
+	if !acceptsAddRemoveFlowUpdates {
+		if err := agent.adapterProxy.UpdateFlowsBulk(device, latestData, groups); err != nil {
+			log.Debugw("update-flow-bulk-error", log.Fields{"id": agent.lastData.Id, "error": err})
+			return err
+		}
+		return nil
+	}
+	// Incremental flow changes accepted
+	var toAdd []*ofp.OfpFlowStats
+	var toDelete []*ofp.OfpFlowStats
+
+	for _, flow := range latestData.Items {
+		if fu.FindFlowById(previousData.Items, flow) == -1 { // did not exist before
+			toAdd = append(toAdd, flow)
+		}
+	}
+	for _, flow := range previousData.Items {
+		if fu.FindFlowById(latestData.Items, flow) == -1 { // does not exist now
+			toDelete = append(toDelete, flow)
+		}
+	}
+	flowChanges := &ofp.FlowChanges{
+		ToAdd:    &voltha.Flows{Items: toAdd},
+		ToRemove: &voltha.Flows{Items: toDelete},
+	}
+	// Send an empty group changes as it would be dealt with a call to groupTableUpdated
+	groupChanges := &ofp.FlowGroupChanges{}
+
+	// Send changes only
+	if err := agent.adapterProxy.UpdateFlowsIncremental(device, flowChanges, groupChanges); err != nil {
+		log.Debugw("update-flow-bulk-error", log.Fields{"id": agent.lastData.Id, "error": err})
+		return err
+	}
+
+	return nil
+}
+
+//groupTableUpdated is the callback after group table has been updated in the model to push them
+//to the adapters
+func (agent *DeviceAgent) groupTableUpdated(args ...interface{}) interface{} {
+	log.Debugw("groupTableUpdated-callback", log.Fields{"argsLen": len(args)})
+
+	agent.lockDevice.Lock()
+	defer agent.lockDevice.Unlock()
+
+	var previousData *voltha.FlowGroups
+	var latestData *voltha.FlowGroups
+
+	var ok bool
+	if previousData, ok = args[0].(*ofp.FlowGroups); !ok {
+		log.Errorw("invalid-args", log.Fields{"args0": args[0]})
+		return nil
+	}
+	if latestData, ok = args[1].(*ofp.FlowGroups); !ok {
+		log.Errorw("invalid-args", log.Fields{"args1": args[1]})
+		return nil
+	}
+
+	// Sanity check - should not happen as this is already handled in logical device agent
+	if reflect.DeepEqual(previousData.Items, latestData.Items) {
+		log.Debugw("group-table-update-not-required", log.Fields{"previous": previousData.Items, "new": latestData.Items})
+		return nil
+	}
+
+	var device *voltha.Device
+	var err error
+	if device, err = agent.getDeviceWithoutLock(); err != nil {
+		log.Errorw("no-device", log.Fields{"id": agent.deviceId, "error": err})
+		return nil
+	}
+	flows := device.Flows
+
+	// Send update to adapters
+	// Check whether the device supports incremental flow changes
+	// Assume false for test
+	acceptsAddRemoveFlowUpdates := false
+	if !acceptsAddRemoveFlowUpdates {
+		if err := agent.adapterProxy.UpdateFlowsBulk(device, flows, latestData); err != nil {
+			log.Debugw("update-flows-bulk-error", log.Fields{"id": agent.lastData.Id, "error": err})
+			return err
+		}
+		return nil
+	}
+
+	// Incremental group changes accepted
+	var toAdd []*ofp.OfpGroupEntry
+	var toDelete []*ofp.OfpGroupEntry
+	var toUpdate []*ofp.OfpGroupEntry
+
+	for _, group := range latestData.Items {
+		if idx := fu.FindGroup(previousData.Items, group.Desc.GroupId); idx == -1 { // did not exist before
+			toAdd = append(toAdd, group)
+		} else { // existed before
+			if previousData.Items[idx].String() != group.String() { // there is a change
+				toUpdate = append(toUpdate, group)
+			}
+		}
+	}
+	for _, group := range previousData.Items {
+		if fu.FindGroup(latestData.Items, group.Desc.GroupId) == -1 { // does not exist now
+			toDelete = append(toDelete, group)
+		}
+	}
+	groupChanges := &ofp.FlowGroupChanges{
+		ToAdd:    &voltha.FlowGroups{Items: toAdd},
+		ToRemove: &voltha.FlowGroups{Items: toDelete},
+		ToUpdate: &voltha.FlowGroups{Items: toUpdate},
+	}
+	// Send an empty flow changes as it should have been dealt with a call to flowTableUpdated
+	flowChanges := &ofp.FlowChanges{}
+
+	// Send changes only
+	if err := agent.adapterProxy.UpdateFlowsIncremental(device, flowChanges, groupChanges); err != nil {
+		log.Debugw("update-incremental-group-error", log.Fields{"id": agent.lastData.Id, "error": err})
+		return err
+	}
+	return nil
+}
+
 // TODO: A generic device update by attribute
 func (agent *DeviceAgent) updateDeviceAttribute(name string, value interface{}) {
 	agent.lockDevice.Lock()
diff --git a/rw_core/core/device_manager.go b/rw_core/core/device_manager.go
index 1063e29..936e945 100644
--- a/rw_core/core/device_manager.go
+++ b/rw_core/core/device_manager.go
@@ -23,6 +23,7 @@
 	"github.com/opencord/voltha-go/db/model"
 	"github.com/opencord/voltha-go/kafka"
 	"github.com/opencord/voltha-go/protos/core_adapter"
+	ofp "github.com/opencord/voltha-go/protos/openflow_13"
 	"github.com/opencord/voltha-go/protos/voltha"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -167,14 +168,22 @@
 	sendResponse(ctx, ch, res)
 }
 
-func (dMgr *DeviceManager) getDevice(id string) (*voltha.Device, error) {
-	log.Debugw("getDevice", log.Fields{"deviceid": id})
+func (dMgr *DeviceManager) GetDevice(id string) (*voltha.Device, error) {
+	log.Debugw("GetDevice", log.Fields{"deviceid": id})
 	if agent := dMgr.getDeviceAgent(id); agent != nil {
 		return agent.getDevice()
 	}
 	return nil, status.Errorf(codes.NotFound, "%s", id)
 }
 
+func (dMgr *DeviceManager) IsRootDevice(id string) (bool, error) {
+	device, err := dMgr.GetDevice(id)
+	if err != nil {
+		return false, err
+	}
+	return device.Root, nil
+}
+
 func (dMgr *DeviceManager) ListDevices() (*voltha.Devices, error) {
 	log.Debug("ListDevices")
 	result := &voltha.Devices{}
@@ -218,6 +227,21 @@
 	}
 }
 
+func (dMgr *DeviceManager) updateFlows(deviceId string, flows []*ofp.OfpFlowStats) error {
+	log.Debugw("updateFlows", log.Fields{"deviceid": deviceId})
+	if agent := dMgr.getDeviceAgent(deviceId); agent != nil {
+		return agent.updateFlows(flows)
+	}
+	return status.Errorf(codes.NotFound, "%s", deviceId)
+}
+
+func (dMgr *DeviceManager) updateGroups(deviceId string, groups []*ofp.OfpGroupEntry) error {
+	if agent := dMgr.getDeviceAgent(deviceId); agent != nil {
+		return agent.updateGroups(groups)
+	}
+	return status.Errorf(codes.NotFound, "%s", deviceId)
+}
+
 func (dMgr *DeviceManager) updatePmConfigs(deviceId string, pmConfigs *voltha.PmConfigs) error {
 	if agent := dMgr.getDeviceAgent(deviceId); agent != nil {
 		return agent.updatePmConfigs(pmConfigs)
@@ -262,7 +286,7 @@
 	log.Debugw("updateChildrenStatus", log.Fields{"parentDeviceid": deviceId, "operStatus": operStatus, "connStatus": connStatus})
 	var parentDevice *voltha.Device
 	var err error
-	if parentDevice, err = dMgr.getDevice(deviceId); err != nil {
+	if parentDevice, err = dMgr.GetDevice(deviceId); err != nil {
 		return status.Errorf(codes.Aborted, "%s", err.Error())
 	}
 	var childDeviceIds []string
@@ -356,11 +380,17 @@
 	return nil
 }
 
-func (dMgr *DeviceManager) deleteLogicalPort(cDevice *voltha.Device) error {
+func (dMgr *DeviceManager) deleteLogicalPort(device *voltha.Device) error {
 	log.Info("deleteLogicalPort")
 	var err error
-	if err = dMgr.logicalDeviceMgr.deleteLogicalPort(nil, cDevice); err != nil {
-		log.Warnw("deleteLogical-port-error", log.Fields{"deviceId": cDevice.Id})
+	// Get the logical port associated with this device
+	var lPortId *voltha.LogicalPortId
+	if lPortId, err = dMgr.logicalDeviceMgr.getLogicalPortId(device); err != nil {
+		log.Warnw("getLogical-port-error", log.Fields{"deviceId": device.Id})
+		return err
+	}
+	if err = dMgr.logicalDeviceMgr.deleteLogicalPort(nil, lPortId); err != nil {
+		log.Warnw("deleteLogical-port-error", log.Fields{"deviceId": device.Id})
 		return err
 	}
 	return nil
@@ -372,7 +402,7 @@
 		// childDevice is the parent device
 		return childDevice
 	}
-	parentDevice, _ := dMgr.getDevice(childDevice.ParentId)
+	parentDevice, _ := dMgr.GetDevice(childDevice.ParentId)
 	return parentDevice
 }
 
@@ -502,7 +532,7 @@
 }
 
 func (dMgr *DeviceManager) GetParentDeviceId(deviceId string) *string {
-	if device, _ := dMgr.getDevice(deviceId); device != nil {
+	if device, _ := dMgr.GetDevice(deviceId); device != nil {
 		log.Infow("GetParentDeviceId", log.Fields{"deviceId": device.Id, "parentId": device.ParentId})
 		return &device.ParentId
 	}
diff --git a/rw_core/core/grpc_nbi_api_handler.go b/rw_core/core/grpc_nbi_api_handler.go
index e036998..ed30bb7 100644
--- a/rw_core/core/grpc_nbi_api_handler.go
+++ b/rw_core/core/grpc_nbi_api_handler.go
@@ -75,11 +75,6 @@
 	return out, nil
 }
 
-func processEnableDevicePort(ctx context.Context, id *voltha.LogicalPortId, ch chan interface{}) {
-	log.Debugw("processEnableDevicePort", log.Fields{"id": id, "test": common.TestModeKeys_api_test.String()})
-	ch <- status.Errorf(100, "%d-%s", 100, "erreur")
-}
-
 func (handler *APIHandler) EnableLogicalDevicePort(ctx context.Context, id *voltha.LogicalPortId) (*empty.Empty, error) {
 	log.Debugw("EnableLogicalDevicePort-request", log.Fields{"id": id, "test": common.TestModeKeys_api_test.String()})
 	if isTestMode(ctx) {
@@ -88,7 +83,7 @@
 	}
 	ch := make(chan interface{})
 	defer close(ch)
-	go processEnableDevicePort(ctx, id, ch)
+	go handler.logicalDeviceMgr.enableLogicalPort(ctx, id, ch)
 	return waitForNilResponseOnSuccess(ctx, ch)
 }
 
@@ -98,7 +93,10 @@
 		out := new(empty.Empty)
 		return out, nil
 	}
-	return nil, errors.New("Unimplemented")
+	ch := make(chan interface{})
+	defer close(ch)
+	go handler.logicalDeviceMgr.disableLogicalPort(ctx, id, ch)
+	return waitForNilResponseOnSuccess(ctx, ch)
 }
 
 func (handler *APIHandler) UpdateLogicalDeviceFlowTable(ctx context.Context, flow *openflow_13.FlowTableUpdate) (*empty.Empty, error) {
@@ -107,7 +105,10 @@
 		out := new(empty.Empty)
 		return out, nil
 	}
-	return nil, errors.New("Unimplemented")
+	ch := make(chan interface{})
+	defer close(ch)
+	go handler.logicalDeviceMgr.updateFlowTable(ctx, flow.Id, flow.FlowMod, ch)
+	return waitForNilResponseOnSuccess(ctx, ch)
 }
 
 func (handler *APIHandler) UpdateLogicalDeviceFlowGroupTable(ctx context.Context, flow *openflow_13.FlowGroupTableUpdate) (*empty.Empty, error) {
@@ -116,13 +117,16 @@
 		out := new(empty.Empty)
 		return out, nil
 	}
-	return nil, errors.New("Unimplemented")
+	ch := make(chan interface{})
+	defer close(ch)
+	go handler.logicalDeviceMgr.updateGroupTable(ctx, flow.Id, flow.GroupMod, ch)
+	return waitForNilResponseOnSuccess(ctx, ch)
 }
 
 // GetDevice must be implemented in the read-only containers - should it also be implemented here?
 func (handler *APIHandler) GetDevice(ctx context.Context, id *voltha.ID) (*voltha.Device, error) {
 	log.Debugw("GetDevice-request", log.Fields{"id": id})
-	return handler.deviceMgr.getDevice(id.Id)
+	return handler.deviceMgr.GetDevice(id.Id)
 }
 
 // GetDevice must be implemented in the read-only containers - should it also be implemented here?
@@ -143,6 +147,12 @@
 	return handler.logicalDeviceMgr.listLogicalDevices()
 }
 
+// ListLogicalDevicePorts must be implemented in the read-only containers - should it also be implemented here?
+func (handler *APIHandler) ListLogicalDevicePorts(ctx context.Context, id *voltha.ID) (*voltha.LogicalPorts, error) {
+	log.Debugw("ListLogicalDevicePorts", log.Fields{"logicaldeviceid": id})
+	return handler.logicalDeviceMgr.ListLogicalDevicePorts(ctx, id.Id)
+}
+
 // CreateDevice creates a new parent device in the data model
 func (handler *APIHandler) CreateDevice(ctx context.Context, device *voltha.Device) (*voltha.Device, error) {
 	log.Debugw("createdevice", log.Fields{"device": *device})
diff --git a/rw_core/core/logical_device_agent.go b/rw_core/core/logical_device_agent.go
index 519a0a1..5c9eced 100644
--- a/rw_core/core/logical_device_agent.go
+++ b/rw_core/core/logical_device_agent.go
@@ -17,6 +17,8 @@
 
 import (
 	"context"
+	"errors"
+	"fmt"
 	"github.com/gogo/protobuf/proto"
 	"github.com/opencord/voltha-go/common/log"
 	"github.com/opencord/voltha-go/db/model"
@@ -28,6 +30,7 @@
 	fu "github.com/opencord/voltha-go/rw_core/utils"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"reflect"
 	"sync"
 )
 
@@ -41,7 +44,10 @@
 	exitChannel       chan int
 	deviceGraph       *graph.DeviceGraph
 	DefaultFlowRules  *fu.DeviceRules
+	flowProxy         *model.Proxy
+	groupProxy        *model.Proxy
 	lockLogicalDevice sync.RWMutex
+	flowDecomposer    *fd.FlowDecomposer
 }
 
 func newLogicalDeviceAgent(id string, device *voltha.Device, ldeviceMgr *LogicalDeviceManager, deviceMgr *DeviceManager,
@@ -53,7 +59,7 @@
 	agent.deviceMgr = deviceMgr
 	agent.clusterDataProxy = cdProxy
 	agent.ldeviceMgr = ldeviceMgr
-	//agent.deviceGraph =
+	agent.flowDecomposer = fd.NewFlowDecomposer(agent.deviceMgr)
 	agent.lockLogicalDevice = sync.RWMutex{}
 	return &agent
 }
@@ -71,6 +77,8 @@
 	ld := &voltha.LogicalDevice{Id: agent.logicalDeviceId, RootDeviceId: agent.rootDeviceId}
 	ld.Desc = (proto.Clone(switchCap.Desc)).(*ofp.OfpDesc)
 	ld.SwitchFeatures = (proto.Clone(switchCap.SwitchFeatures)).(*ofp.OfpSwitchFeatures)
+	ld.Flows = &ofp.Flows{Items: nil}
+	ld.FlowGroups = &ofp.FlowGroups{Items: nil}
 
 	//Add logical ports to the logical device based on the number of NNI ports discovered
 	//First get the default port capability - TODO:  each NNI port may have different capabilities,
@@ -82,14 +90,18 @@
 	}
 	var portCap *ca.PortCapability
 	for _, port := range nniPorts.Items {
-		log.Infow("NNI PORTS", log.Fields{"NNI": port})
+		log.Infow("!!!!!!!NNI PORTS", log.Fields{"NNI": port})
 		if portCap, err = agent.deviceMgr.getPortCapability(ctx, agent.rootDeviceId, port.PortNo); err != nil {
 			log.Errorw("error-creating-logical-device", log.Fields{"error": err})
 			return err
 		}
-
+		portCap.Port.RootPort = true
 		lp := (proto.Clone(portCap.Port)).(*voltha.LogicalPort)
 		lp.DeviceId = agent.rootDeviceId
+		lp.Id = fmt.Sprintf("nni-%d", port.PortNo)
+		lp.OfpPort.PortNo = port.PortNo
+		lp.OfpPort.Name = portCap.Port.Id
+		lp.DevicePortNo = port.PortNo
 		ld.Ports = append(ld.Ports, lp)
 	}
 	agent.lockLogicalDevice.Lock()
@@ -101,6 +113,16 @@
 		log.Debugw("logicaldevice-created", log.Fields{"logicaldeviceId": agent.logicalDeviceId})
 	}
 
+	agent.flowProxy = agent.clusterDataProxy.Root.GetProxy(
+		fmt.Sprintf("/logical_devices/%s/flows", agent.logicalDeviceId),
+		false)
+	agent.groupProxy = agent.clusterDataProxy.Root.GetProxy(
+		fmt.Sprintf("/logical_devices/%s/flow_groups", agent.logicalDeviceId),
+		false)
+
+	agent.flowProxy.RegisterCallback(model.POST_UPDATE, agent.flowTableUpdated)
+	//agent.groupProxy.RegisterCallback(model.POST_UPDATE, agent.groupTableUpdated)
+
 	return nil
 }
 
@@ -119,9 +141,9 @@
 	log.Info("logical_device-agent-stopped")
 }
 
-// getLogicalDevice locks the logical device model and then retrieves the latest logical device information
-func (agent *LogicalDeviceAgent) getLogicalDevice() (*voltha.LogicalDevice, error) {
-	log.Debug("getLogicalDevice")
+// GetLogicalDevice locks the logical device model and then retrieves the latest logical device information
+func (agent *LogicalDeviceAgent) GetLogicalDevice() (*voltha.LogicalDevice, error) {
+	log.Debug("GetLogicalDevice")
 	agent.lockLogicalDevice.Lock()
 	defer agent.lockLogicalDevice.Unlock()
 	logicalDevice := agent.clusterDataProxy.Get("/logical_devices/"+agent.logicalDeviceId, 1, false, "")
@@ -132,6 +154,45 @@
 	return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceId)
 }
 
+func (agent *LogicalDeviceAgent) ListLogicalDevicePorts() (*voltha.LogicalPorts, error) {
+	log.Debug("!!!!!ListLogicalDevicePorts")
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+	logicalDevice := agent.clusterDataProxy.Get("/logical_devices/"+agent.logicalDeviceId, 1, false, "")
+	if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
+		lPorts := make([]*voltha.LogicalPort, 0)
+		for _, port := range lDevice.Ports {
+			lPorts = append(lPorts, proto.Clone(port).(*voltha.LogicalPort))
+		}
+		return &voltha.LogicalPorts{Items: lPorts}, nil
+	}
+	return nil, status.Errorf(codes.NotFound, "logical_device-%s", agent.logicalDeviceId)
+}
+
+// listFlows locks the logical device model and then retrieves the latest flow information
+func (agent *LogicalDeviceAgent) listFlows() []*ofp.OfpFlowStats {
+	log.Debug("listFlows")
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+	logicalDevice := agent.clusterDataProxy.Get("/logical_devices/"+agent.logicalDeviceId, 1, false, "")
+	if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
+		return lDevice.Flows.Items
+	}
+	return nil
+}
+
+// listFlowGroups locks the logical device model and then retrieves the latest flow groups information
+func (agent *LogicalDeviceAgent) listFlowGroups() []*ofp.OfpGroupEntry {
+	log.Debug("listFlowGroups")
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+	logicalDevice := agent.clusterDataProxy.Get("/logical_devices/"+agent.logicalDeviceId, 1, false, "")
+	if lDevice, ok := logicalDevice.(*voltha.LogicalDevice); ok {
+		return lDevice.FlowGroups.Items
+	}
+	return nil
+}
+
 // getLogicalDeviceWithoutLock retrieves a logical device from the model without locking it.   This is used only by
 // functions that have already acquired the logical device lock to the model
 func (agent *LogicalDeviceAgent) getLogicalDeviceWithoutLock() (*voltha.LogicalDevice, error) {
@@ -145,12 +206,20 @@
 }
 
 // addUNILogicalPort creates a UNI port on the logical device that represents a child device
-func (agent *LogicalDeviceAgent) addUNILogicalPort(ctx context.Context, childDevice *voltha.Device, portNo uint32) error {
+func (agent *LogicalDeviceAgent) addUNILogicalPort(ctx context.Context, childDevice *voltha.Device) error {
 	log.Infow("addUNILogicalPort-start", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
 	// Build the logical device based on information retrieved from the device adapter
 	var portCap *ca.PortCapability
 	var err error
-	if portCap, err = agent.deviceMgr.getPortCapability(ctx, childDevice.Id, portNo); err != nil {
+
+	//Get UNI port number
+	var uniPort uint32
+	for _, port := range childDevice.Ports {
+		if port.Type == voltha.Port_ETHERNET_UNI {
+			uniPort = port.PortNo
+		}
+	}
+	if portCap, err = agent.deviceMgr.getPortCapability(ctx, childDevice.Id, uniPort); err != nil {
 		log.Errorw("error-creating-logical-port", log.Fields{"error": err})
 		return err
 	}
@@ -160,7 +229,16 @@
 	if ldevice, err := agent.getLogicalDeviceWithoutLock(); err != nil {
 		return status.Error(codes.NotFound, agent.logicalDeviceId)
 	} else {
+		log.Infow("!!!!!!!!!!!ADDING-UNI", log.Fields{"deviceId": childDevice.Id})
 		cloned := proto.Clone(ldevice).(*voltha.LogicalDevice)
+		portCap.Port.RootPort = false
+		//TODO: For now use the channel id assigned by the OLT as logical port number
+		lPortNo := childDevice.ProxyAddress.ChannelId
+		portCap.Port.Id = fmt.Sprintf("uni-%d", lPortNo)
+		portCap.Port.OfpPort.PortNo = lPortNo
+		portCap.Port.OfpPort.Name = portCap.Port.Id
+		portCap.Port.DeviceId = childDevice.Id
+		portCap.Port.DevicePortNo = uniPort
 		lp := proto.Clone(portCap.Port).(*voltha.LogicalPort)
 		lp.DeviceId = childDevice.Id
 		cloned.Ports = append(cloned.Ports, lp)
@@ -178,19 +256,382 @@
 	return nil
 }
 
-// deleteLogicalPort removes the logical port associated with a child device
-func (agent *LogicalDeviceAgent) deleteLogicalPort(device *voltha.Device) error {
+//updateFlowTable updates the flow table of that logical device
+func (agent *LogicalDeviceAgent) updateFlowTable(ctx context.Context, flow *ofp.OfpFlowMod) error {
+	log.Debug("updateFlowTable")
+	if flow == nil {
+		return nil
+	}
+	switch flow.GetCommand() {
+	case ofp.OfpFlowModCommand_OFPFC_ADD:
+		return agent.flowAdd(flow)
+	case ofp.OfpFlowModCommand_OFPFC_DELETE:
+		return agent.flowDelete(flow)
+	case ofp.OfpFlowModCommand_OFPFC_DELETE_STRICT:
+		return agent.flowDeleteStrict(flow)
+	case ofp.OfpFlowModCommand_OFPFC_MODIFY:
+		return agent.flowModify(flow)
+	case ofp.OfpFlowModCommand_OFPFC_MODIFY_STRICT:
+		return agent.flowModifyStrict(flow)
+	}
+	return status.Errorf(codes.Internal,
+		"unhandled-command: lDeviceId:%s, command:%s", agent.logicalDeviceId, flow.GetCommand())
+}
+
+//updateGroupTable updates the group table of that logical device
+func (agent *LogicalDeviceAgent) updateGroupTable(ctx context.Context, groupMod *ofp.OfpGroupMod) error {
+	log.Debug("updateGroupTable")
+	if groupMod == nil {
+		return nil
+	}
+	switch groupMod.GetCommand() {
+	case ofp.OfpGroupModCommand_OFPGC_ADD:
+		return agent.groupAdd(groupMod)
+	case ofp.OfpGroupModCommand_OFPGC_DELETE:
+		return agent.groupDelete(groupMod)
+	case ofp.OfpGroupModCommand_OFPGC_MODIFY:
+		return agent.groupModify(groupMod)
+	}
+	return status.Errorf(codes.Internal,
+		"unhandled-command: lDeviceId:%s, command:%s", agent.logicalDeviceId, groupMod.GetCommand())
+}
+
+//updateFlowsWithoutLock updates the flows in the logical device without locking the logical device.  This function
+//must only be called by a function that is holding the lock on the logical device
+func (agent *LogicalDeviceAgent) updateFlowsWithoutLock(flows []*ofp.OfpFlowStats) error {
+	if ldevice, err := agent.getLogicalDeviceWithoutLock(); err != nil {
+		return status.Error(codes.NotFound, agent.logicalDeviceId)
+	} else {
+		flowsCloned := make([]*ofp.OfpFlowStats, len(flows))
+		copy(flowsCloned, flows)
+		ldevice.Flows.Items = flowsCloned
+		return agent.updateLogicalDeviceWithoutLock(ldevice)
+	}
+}
+
+//updateFlowGroupsWithoutLock updates the flows in the logical device without locking the logical device.  This function
+//must only be called by a function that is holding the lock on the logical device
+func (agent *LogicalDeviceAgent) updateFlowGroupsWithoutLock(groups []*ofp.OfpGroupEntry) error {
+	if ldevice, err := agent.getLogicalDeviceWithoutLock(); err != nil {
+		return status.Error(codes.NotFound, agent.logicalDeviceId)
+	} else {
+		groupsCloned := make([]*ofp.OfpGroupEntry, len(groups))
+		copy(groupsCloned, groups)
+		ldevice.FlowGroups.Items = groupsCloned
+		return agent.updateLogicalDeviceWithoutLock(ldevice)
+	}
+}
+
+//flowAdd adds a flow to the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowAdd(mod *ofp.OfpFlowMod) error {
+	log.Debug("flowAdd")
+	if mod == nil {
+		return nil
+	}
 	agent.lockLogicalDevice.Lock()
 	defer agent.lockLogicalDevice.Unlock()
+
+	var lDevice *voltha.LogicalDevice
+	var err error
+	if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+		log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+		return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+	}
+
+	var flows []*ofp.OfpFlowStats
+	if lDevice.Flows != nil && lDevice.Flows.Items != nil {
+		flows = lDevice.Flows.Items
+	}
+
+	oldData := proto.Clone(lDevice.Flows).(*voltha.Flows)
+	changed := false
+	checkOverlap := (mod.Flags & uint32(ofp.OfpFlowModFlags_OFPFF_CHECK_OVERLAP)) != 0
+	if checkOverlap {
+		if overlapped := fu.FindOverlappingFlows(flows, mod); len(overlapped) != 0 {
+			//	TODO:  should this error be notified other than being logged?
+			log.Warnw("overlapped-flows", log.Fields{"logicaldeviceId": agent.logicalDeviceId})
+		} else {
+			//	Add flow
+			flow := fd.FlowStatsEntryFromFlowModMessage(mod)
+			flows = append(flows, flow)
+			changed = true
+		}
+	} else {
+		flow := fd.FlowStatsEntryFromFlowModMessage(mod)
+		idx := fu.FindFlows(flows, flow)
+		if idx >= 0 {
+			oldFlow := flows[idx]
+			if (mod.Flags & uint32(ofp.OfpFlowModFlags_OFPFF_RESET_COUNTS)) != 0 {
+				flow.ByteCount = oldFlow.ByteCount
+				flow.PacketCount = oldFlow.PacketCount
+			}
+			flows[idx] = flow
+		} else {
+			flows = append(flows, flow)
+		}
+		changed = true
+	}
+	if changed {
+		//	Update model
+		if lDevice.Flows == nil {
+			lDevice.Flows = &ofp.Flows{}
+		}
+		lDevice.Flows.Items = flows
+		if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+			log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+			return err
+		}
+	}
+	// For now, force the callback to occur
+	go agent.flowTableUpdated(oldData, lDevice.Flows)
+	return nil
+}
+
+//flowDelete deletes a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowDelete(mod *ofp.OfpFlowMod) error {
+	log.Debug("flowDelete")
+	if mod == nil {
+		return nil
+	}
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
+	var lDevice *voltha.LogicalDevice
+	var err error
+	if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+		log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+		return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+	}
+	flows := lDevice.Flows.Items
+
+	//build a list of what to keep vs what to delete
+	toKeep := make([]*ofp.OfpFlowStats, 0)
+	for _, f := range flows {
+		if !fu.FlowMatchesMod(f, mod) {
+			toKeep = append(toKeep, f)
+		}
+	}
+
+	//Update flows
+	if len(toKeep) < len(flows) {
+		lDevice.Flows.Items = toKeep
+		if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+			log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+			return err
+		}
+	}
+
+	//TODO: send announcement on delete
+	return nil
+}
+
+//flowStatsDelete deletes a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowStatsDelete(flow *ofp.OfpFlowStats) error {
+	log.Debug("flowStatsDelete")
+	if flow == nil {
+		return nil
+	}
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
+	var lDevice *voltha.LogicalDevice
+	var err error
+	if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+		log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+		return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+	}
+	flows := lDevice.Flows.Items
+
+	//build a list of what to keep vs what to delete
+	toKeep := make([]*ofp.OfpFlowStats, 0)
+	for _, f := range flows {
+		if !fu.FlowMatch(f, flow) {
+			toKeep = append(toKeep, f)
+		}
+	}
+
+	//Update flows
+	if len(toKeep) < len(flows) {
+		lDevice.Flows.Items = toKeep
+		if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+			log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+			return err
+		}
+	}
+	return nil
+}
+
+//flowDeleteStrict deletes a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowDeleteStrict(mod *ofp.OfpFlowMod) error {
+	log.Debug("flowDeleteStrict")
+	if mod == nil {
+		return nil
+	}
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
+	var lDevice *voltha.LogicalDevice
+	var err error
+	if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+		log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+		return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+	}
+	flows := lDevice.Flows.Items
+	changed := false
+	flow := fd.FlowStatsEntryFromFlowModMessage(mod)
+	idx := fu.FindFlows(flows, flow)
+	if idx >= 0 {
+		flows = append(flows[:idx], flows[idx+1:]...)
+		changed = true
+	} else {
+		return errors.New(fmt.Sprintf("Cannot delete flow - %s", flow))
+	}
+
+	if changed {
+		lDevice.Flows.Items = flows
+		if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+			log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+			return err
+		}
+	}
+
+	return nil
+}
+
+//flowModify modifies a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowModify(mod *ofp.OfpFlowMod) error {
+	return errors.New("flowModify not implemented")
+}
+
+//flowModifyStrict deletes a flow from the flow table of that logical device
+func (agent *LogicalDeviceAgent) flowModifyStrict(mod *ofp.OfpFlowMod) error {
+	return errors.New("flowModifyStrict not implemented")
+}
+
+func (agent *LogicalDeviceAgent) groupAdd(groupMod *ofp.OfpGroupMod) error {
+	log.Debug("groupAdd")
+	if groupMod == nil {
+		return nil
+	}
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
+	var lDevice *voltha.LogicalDevice
+	var err error
+	if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+		log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+		return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+	}
+	groups := lDevice.FlowGroups.Items
+	oldData := proto.Clone(lDevice.FlowGroups).(*voltha.FlowGroups)
+	if fu.FindGroup(groups, groupMod.GroupId) == -1 {
+		groups = append(groups, fd.GroupEntryFromGroupMod(groupMod))
+		lDevice.FlowGroups.Items = groups
+		if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+			log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+			return err
+		}
+	} else {
+		return errors.New(fmt.Sprintf("Groups %d already present", groupMod.GroupId))
+	}
+	// For now, force the callback to occur
+	go agent.groupTableUpdated(oldData, lDevice.FlowGroups)
+	return nil
+}
+
+func (agent *LogicalDeviceAgent) groupDelete(groupMod *ofp.OfpGroupMod) error {
+	log.Debug("groupDelete")
+	if groupMod == nil {
+		return nil
+	}
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
+	var lDevice *voltha.LogicalDevice
+	var err error
+	if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+		log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+		return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+	}
+	groups := lDevice.FlowGroups.Items
+	flows := lDevice.Flows.Items
+	groupsChanged := false
+	flowsChanged := false
+	groupId := groupMod.GroupId
+	if groupId == uint32(ofp.OfpGroup_OFPG_ALL) {
+		//TODO we must delete all flows that point to this group and
+		//signal controller as requested by flow's flag
+		groups = []*ofp.OfpGroupEntry{}
+		groupsChanged = true
+	} else {
+		if idx := fu.FindGroup(groups, groupId); idx == -1 {
+			return nil // Valid case
+		} else {
+			flowsChanged, flows = fu.FlowsDeleteByGroupId(flows, groupId)
+			groups = append(groups[:idx], groups[idx+1:]...)
+			groupsChanged = true
+		}
+	}
+	if groupsChanged || flowsChanged {
+		lDevice.FlowGroups.Items = groups
+		lDevice.Flows.Items = flows
+		if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+			log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+			return err
+		}
+	}
+	return nil
+}
+
+func (agent *LogicalDeviceAgent) groupModify(groupMod *ofp.OfpGroupMod) error {
+	log.Debug("groupModify")
+	if groupMod == nil {
+		return nil
+	}
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
+	var lDevice *voltha.LogicalDevice
+	var err error
+	if lDevice, err = agent.getLogicalDeviceWithoutLock(); err != nil {
+		log.Errorw("no-logical-device-present", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+		return errors.New(fmt.Sprintf("no-logical-device-present:%s", agent.logicalDeviceId))
+	}
+	groups := lDevice.FlowGroups.Items
+	groupsChanged := false
+	groupId := groupMod.GroupId
+	if idx := fu.FindGroup(groups, groupId); idx == -1 {
+		return errors.New(fmt.Sprintf("group-absent:%s", groupId))
+	} else {
+		//replace existing group entry with new group definition
+		groupEntry := fd.GroupEntryFromGroupMod(groupMod)
+		groups[idx] = groupEntry
+		groupsChanged = true
+	}
+	if groupsChanged {
+		lDevice.FlowGroups.Items = groups
+		if err := agent.updateLogicalDeviceWithoutLock(lDevice); err != nil {
+			log.Errorw("Cannot-update-logical-group", log.Fields{"logicalDeviceId": agent.logicalDeviceId})
+			return err
+		}
+	}
+	return nil
+}
+
+// deleteLogicalPort removes the logical port
+func (agent *LogicalDeviceAgent) deleteLogicalPort(lPort *voltha.LogicalPort) error {
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
 	// Get the most up to date logical device
 	var logicaldevice *voltha.LogicalDevice
 	if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
-		log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "deviceId": device.Id})
+		log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "logicalPortId": lPort.Id})
 		return nil
 	}
 	index := -1
 	for i, logicalPort := range logicaldevice.Ports {
-		if logicalPort.DeviceId == device.Id {
+		if logicalPort.Id == lPort.Id {
 			index = i
 			break
 		}
@@ -205,6 +646,58 @@
 	return nil
 }
 
+// enableLogicalPort enables the logical port
+func (agent *LogicalDeviceAgent) enableLogicalPort(lPort *voltha.LogicalPort) error {
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
+	// Get the most up to date logical device
+	var logicaldevice *voltha.LogicalDevice
+	if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
+		log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "logicalPortId": lPort.Id})
+		return nil
+	}
+	index := -1
+	for i, logicalPort := range logicaldevice.Ports {
+		if logicalPort.Id == lPort.Id {
+			index = i
+			break
+		}
+	}
+	if index >= 0 {
+		logicaldevice.Ports[index].OfpPort.Config = logicaldevice.Ports[index].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+		return agent.updateLogicalDeviceWithoutLock(logicaldevice)
+	}
+	//TODO:  Trigger subsequent actions on the device
+	return nil
+}
+
+// disableLogicalPort disabled the logical port
+func (agent *LogicalDeviceAgent) disableLogicalPort(lPort *voltha.LogicalPort) error {
+	agent.lockLogicalDevice.Lock()
+	defer agent.lockLogicalDevice.Unlock()
+
+	// Get the most up to date logical device
+	var logicaldevice *voltha.LogicalDevice
+	if logicaldevice, _ = agent.getLogicalDeviceWithoutLock(); logicaldevice == nil {
+		log.Debugw("no-logical-device", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "logicalPortId": lPort.Id})
+		return nil
+	}
+	index := -1
+	for i, logicalPort := range logicaldevice.Ports {
+		if logicalPort.Id == lPort.Id {
+			index = i
+			break
+		}
+	}
+	if index >= 0 {
+		logicaldevice.Ports[index].OfpPort.Config = (logicaldevice.Ports[index].OfpPort.Config & ^uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)) | uint32(ofp.OfpPortConfig_OFPPC_PORT_DOWN)
+		return agent.updateLogicalDeviceWithoutLock(logicaldevice)
+	}
+	//TODO:  Trigger subsequent actions on the device
+	return nil
+}
+
 func isNNIPort(portNo uint32, nniPortsNo []uint32) bool {
 	for _, pNo := range nniPortsNo {
 		if pNo == portNo {
@@ -215,7 +708,9 @@
 }
 
 func (agent *LogicalDeviceAgent) getPreCalculatedRoute(ingress, egress uint32) []graph.RouteHop {
+	log.Debugw("ROUTE", log.Fields{"len": len(agent.deviceGraph.Routes)})
 	for routeLink, route := range agent.deviceGraph.Routes {
+		log.Debugw("ROUTELINKS", log.Fields{"ingress": ingress, "egress": egress, "routelink": routeLink})
 		if ingress == routeLink.Ingress && egress == routeLink.Egress {
 			return route
 		}
@@ -224,9 +719,7 @@
 	return nil
 }
 
-func (agent *LogicalDeviceAgent) GetRoute(ingressPortNo *uint32, egressPortNo *uint32) []graph.RouteHop {
-	agent.lockLogicalDevice.Lock()
-	defer agent.lockLogicalDevice.Unlock()
+func (agent *LogicalDeviceAgent) GetRoute(ingressPortNo uint32, egressPortNo uint32) []graph.RouteHop {
 	log.Debugw("getting-route", log.Fields{"ingress-port": ingressPortNo, "egress-port": egressPortNo})
 	// Get the updated logical device
 	var ld *ca.LogicalDevice
@@ -245,10 +738,12 @@
 		log.Errorw("no-nni-ports", log.Fields{"LogicalDeviceId": ld.Id})
 		return nil
 	}
+	// Note: A port value of 0 is equivalent to a nil port
+
 	//	Consider different possibilities
-	if egressPortNo != nil && ((*egressPortNo & 0x7fffffff) == uint32(ofp.OfpPortNo_OFPP_CONTROLLER)) {
+	if egressPortNo != 0 && ((egressPortNo & 0x7fffffff) == uint32(ofp.OfpPortNo_OFPP_CONTROLLER)) {
 		log.Debugw("controller-flow", log.Fields{"ingressPortNo": ingressPortNo, "egressPortNo": egressPortNo, "nniPortsNo": nniLogicalPortsNo})
-		if isNNIPort(*ingressPortNo, nniLogicalPortsNo) {
+		if isNNIPort(ingressPortNo, nniLogicalPortsNo) {
 			log.Debug("returning-half-route")
 			//This is a trap on the NNI Port
 			//Return a 'half' route to make the flow decomposer logic happy
@@ -263,13 +758,13 @@
 			return nil
 		}
 		//treat it as if the output port is the first NNI of the OLT
-		egressPortNo = &nniLogicalPortsNo[0]
+		egressPortNo = nniLogicalPortsNo[0]
 	}
 	//If ingress port is not specified (nil), it may be a wildcarded
 	//route if egress port is OFPP_CONTROLLER or a nni logical port,
 	//in which case we need to create a half-route where only the egress
 	//hop is filled, the first hop is nil
-	if ingressPortNo == nil && isNNIPort(*egressPortNo, nniLogicalPortsNo) {
+	if ingressPortNo == 0 && isNNIPort(egressPortNo, nniLogicalPortsNo) {
 		// We can use the 2nd hop of any upstream route, so just find the first upstream:
 		for routeLink, route := range agent.deviceGraph.Routes {
 			if isNNIPort(routeLink.Egress, nniLogicalPortsNo) {
@@ -282,9 +777,9 @@
 		return nil
 	}
 	//If egress port is not specified (nil), we can also can return a "half" route
-	if egressPortNo == nil {
+	if egressPortNo == 0 {
 		for routeLink, route := range agent.deviceGraph.Routes {
-			if routeLink.Ingress == *ingressPortNo {
+			if routeLink.Ingress == ingressPortNo {
 				routes = append(routes, route[0])
 				routes = append(routes, graph.RouteHop{})
 				return routes
@@ -295,14 +790,14 @@
 	}
 
 	//	Return the pre-calculated route
-	return agent.getPreCalculatedRoute(*ingressPortNo, *egressPortNo)
+	return agent.getPreCalculatedRoute(ingressPortNo, egressPortNo)
 }
 
 // updateRoutes updates the device routes whenever there is a device or port changes relevant to this
 // logical device.   TODO: Add more heuristics to this process to update the routes where a change has occurred
 // instead of rebuilding the entire set of routes
 func (agent *LogicalDeviceAgent) updateRoutes() {
-	if ld, err := agent.getLogicalDevice(); err == nil {
+	if ld, err := agent.GetLogicalDevice(); err == nil {
 		agent.deviceGraph.ComputeRoutes(ld.Ports)
 	}
 }
@@ -315,7 +810,7 @@
 	fg := fu.NewFlowsAndGroups()
 	var device *voltha.Device
 	var err error
-	if device, err = agent.deviceMgr.getDevice(deviceId); err != nil {
+	if device, err = agent.deviceMgr.GetDevice(deviceId); err != nil {
 		return fg
 	}
 	//set the upstream and downstream ports
@@ -343,6 +838,7 @@
 		},
 		Actions: []*ofp.OfpAction{
 			fd.SetField(fd.VlanVid(uint32(ofp.OfpVlanId_OFPVID_PRESENT) | device.Vlan)),
+			fd.Output(upstreamPorts[0].PortNo),
 		},
 	}
 	fg.AddFlow(fd.MkFlowStat(fa))
@@ -381,7 +877,7 @@
 	rules := fu.NewDeviceRules()
 	var ld *voltha.LogicalDevice
 	var err error
-	if ld, err = agent.getLogicalDevice(); err != nil {
+	if ld, err = agent.GetLogicalDevice(); err != nil {
 		log.Warnw("no-logical-device", log.Fields{"logicaldeviceId": agent.logicalDeviceId})
 		return rules
 	}
@@ -401,11 +897,11 @@
 	// Get latest
 	var lDevice *voltha.LogicalDevice
 	var err error
-	if lDevice, err = agent.getLogicalDevice(); err != nil {
+	if lDevice, err = agent.GetLogicalDevice(); err != nil {
 		return fu.NewDeviceRules()
 	}
 	if agent.DefaultFlowRules == nil { // Nothing setup yet
-		agent.deviceGraph = graph.NewDeviceGraph(agent.deviceMgr.getDevice)
+		agent.deviceGraph = graph.NewDeviceGraph(agent.deviceMgr.GetDevice)
 		agent.deviceGraph.ComputeRoutes(lDevice.Ports)
 		agent.DefaultFlowRules = agent.generateDefaultRules()
 	}
@@ -418,7 +914,7 @@
 	if len(excludePort) == 1 {
 		exclPort = excludePort[0]
 	}
-	if lDevice, _ := agent.getLogicalDevice(); lDevice != nil {
+	if lDevice, _ := agent.GetLogicalDevice(); lDevice != nil {
 		for _, port := range lDevice.Ports {
 			if port.OfpPort.PortNo != exclPort {
 				lPorts = append(lPorts, port.OfpPort.PortNo)
@@ -427,3 +923,103 @@
 	}
 	return lPorts
 }
+
+func (agent *LogicalDeviceAgent) GetDeviceGraph() *graph.DeviceGraph {
+	return agent.deviceGraph
+}
+
+//setupDeviceGraph creates the device graph if not done already
+func (agent *LogicalDeviceAgent) setupDeviceGraph() {
+	if agent.deviceGraph == nil {
+		agent.deviceGraph = graph.NewDeviceGraph(agent.deviceMgr.GetDevice)
+		agent.updateRoutes()
+	}
+}
+
+func (agent *LogicalDeviceAgent) flowTableUpdated(args ...interface{}) interface{} {
+	log.Debugw("flowTableUpdated-callback", log.Fields{"argsLen": len(args)})
+
+	//agent.lockLogicalDevice.Lock()
+	//defer agent.lockLogicalDevice.Unlock()
+
+	var previousData *ofp.Flows
+	var latestData *ofp.Flows
+
+	var ok bool
+	if previousData, ok = args[0].(*ofp.Flows); !ok {
+		log.Errorw("invalid-args", log.Fields{"args0": args[0]})
+	}
+	if latestData, ok = args[1].(*ofp.Flows); !ok {
+		log.Errorw("invalid-args", log.Fields{"args1": args[1]})
+	}
+
+	if reflect.DeepEqual(previousData.Items, latestData.Items) {
+		log.Debug("flow-update-not-required")
+		return nil
+	}
+
+	// Ensure the device graph has been setup
+	agent.setupDeviceGraph()
+
+	var groups *ofp.FlowGroups
+	lDevice, _ := agent.getLogicalDeviceWithoutLock()
+	groups = lDevice.FlowGroups
+	log.Debugw("flowsinfo", log.Fields{"flows": latestData, "groups": groups})
+	//groupsIf := agent.groupProxy.Get("/", 1, false, "")
+	//if groups, ok = groupsIf.(*ofp.FlowGroups); !ok {
+	//	log.Errorw("cannot-retrieve-groups", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "group": groupsIf})
+	//	//return errors.New("cannot-retrieve-groups")
+	//	groups = &ofp.FlowGroups{Items:nil}
+	//}
+	deviceRules := agent.flowDecomposer.DecomposeRules(agent, *latestData, *groups)
+	log.Debugw("rules", log.Fields{"rules": deviceRules.String()})
+	for deviceId, value := range deviceRules.GetRules() {
+		agent.deviceMgr.updateFlows(deviceId, value.ListFlows())
+		agent.deviceMgr.updateGroups(deviceId, value.ListGroups())
+	}
+	return nil
+}
+
+func (agent *LogicalDeviceAgent) groupTableUpdated(args ...interface{}) interface{} {
+	log.Debugw("groupTableUpdated-callback", log.Fields{"argsLen": len(args)})
+
+	//agent.lockLogicalDevice.Lock()
+	//defer agent.lockLogicalDevice.Unlock()
+
+	var previousData *ofp.FlowGroups
+	var latestData *ofp.FlowGroups
+
+	var ok bool
+	if previousData, ok = args[0].(*ofp.FlowGroups); !ok {
+		log.Errorw("invalid-args", log.Fields{"args0": args[0]})
+	}
+	if latestData, ok = args[1].(*ofp.FlowGroups); !ok {
+		log.Errorw("invalid-args", log.Fields{"args1": args[1]})
+	}
+
+	if reflect.DeepEqual(previousData.Items, latestData.Items) {
+		log.Debug("flow-update-not-required")
+		return nil
+	}
+
+	// Ensure the device graph has been setup
+	agent.setupDeviceGraph()
+
+	var flows *ofp.Flows
+	lDevice, _ := agent.getLogicalDeviceWithoutLock()
+	flows = lDevice.Flows
+	log.Debugw("groupsinfo", log.Fields{"groups": latestData, "flows": flows})
+	//flowsIf := agent.flowProxy.Get("/", 1, false, "")
+	//if flows, ok = flowsIf.(*ofp.Flows); !ok {
+	//	log.Errorw("cannot-retrieve-flows", log.Fields{"logicalDeviceId": agent.logicalDeviceId, "flows": flows})
+	//	//return errors.New("cannot-retrieve-groups")
+	//	flows = &ofp.Flows{Items:nil}
+	//}
+	deviceRules := agent.flowDecomposer.DecomposeRules(agent, *flows, *latestData)
+	log.Debugw("rules", log.Fields{"rules": deviceRules.String()})
+	for deviceId, value := range deviceRules.GetRules() {
+		agent.deviceMgr.updateFlows(deviceId, value.ListFlows())
+		agent.deviceMgr.updateGroups(deviceId, value.ListGroups())
+	}
+	return nil
+}
diff --git a/rw_core/core/logical_device_manager.go b/rw_core/core/logical_device_manager.go
index 8f8548a..9d365aa 100644
--- a/rw_core/core/logical_device_manager.go
+++ b/rw_core/core/logical_device_manager.go
@@ -21,6 +21,7 @@
 	"github.com/opencord/voltha-go/common/log"
 	"github.com/opencord/voltha-go/db/model"
 	"github.com/opencord/voltha-go/kafka"
+	"github.com/opencord/voltha-go/protos/openflow_13"
 	"github.com/opencord/voltha-go/protos/voltha"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
@@ -60,6 +61,18 @@
 	log.Info("logical-device-manager-stopped")
 }
 
+func sendAPIResponse(ctx context.Context, ch chan interface{}, result interface{}) {
+	if ctx.Err() == nil {
+		// Returned response only of the ctx has not been cancelled/timeout/etc
+		// Channel is automatically closed when a context is Done
+		ch <- result
+		log.Debugw("sendResponse", log.Fields{"result": result})
+	} else {
+		// Should the transaction be reverted back?
+		log.Debugw("sendResponse-context-error", log.Fields{"context-error": ctx.Err()})
+	}
+}
+
 func (ldMgr *LogicalDeviceManager) addLogicalDeviceAgentToMap(agent *LogicalDeviceAgent) {
 	ldMgr.lockLogicalDeviceAgentsMap.Lock()
 	defer ldMgr.lockLogicalDeviceAgentsMap.Unlock()
@@ -83,11 +96,11 @@
 	delete(ldMgr.logicalDeviceAgents, logicalDeviceId)
 }
 
-// getLogicalDevice provides a cloned most up to date logical device
+// GetLogicalDevice provides a cloned most up to date logical device
 func (ldMgr *LogicalDeviceManager) getLogicalDevice(id string) (*voltha.LogicalDevice, error) {
 	log.Debugw("getlogicalDevice", log.Fields{"logicaldeviceid": id})
 	if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
-		return agent.getLogicalDevice()
+		return agent.GetLogicalDevice()
 	}
 	return nil, status.Errorf(codes.NotFound, "%s", id)
 }
@@ -98,7 +111,7 @@
 	ldMgr.lockLogicalDeviceAgentsMap.Lock()
 	defer ldMgr.lockLogicalDeviceAgentsMap.Unlock()
 	for _, agent := range ldMgr.logicalDeviceAgents {
-		if lDevice, err := agent.getLogicalDevice(); err == nil {
+		if lDevice, err := agent.GetLogicalDevice(); err == nil {
 			result.Items = append(result.Items, lDevice)
 		}
 	}
@@ -164,20 +177,67 @@
 	return nil, status.Errorf(codes.NotFound, "%s", device.Id)
 }
 
-// DeleteLogicalDevice removes the logical port associated with a child device
-func (ldMgr *LogicalDeviceManager) deleteLogicalPort(ctx context.Context, device *voltha.Device) error {
-	log.Debugw("deleting-logical-port", log.Fields{"deviceId": device.Id})
+func (ldMgr *LogicalDeviceManager) getLogicalPortId(device *voltha.Device) (*voltha.LogicalPortId, error) {
+	// Get the logical device where this device is attached
+	var lDeviceId *string
+	var err error
+	if lDeviceId, err = ldMgr.getLogicalDeviceId(device); err != nil {
+		return nil, err
+	}
+	var lDevice *voltha.LogicalDevice
+	if lDevice, err = ldMgr.getLogicalDevice(*lDeviceId); err != nil {
+		return nil, err
+	}
+	// Go over list of ports
+	for _, port := range lDevice.Ports {
+		if port.DeviceId == device.Id {
+			return &voltha.LogicalPortId{Id: *lDeviceId, PortId: port.Id}, nil
+		}
+	}
+	return nil, status.Errorf(codes.NotFound, "%s", device.Id)
+}
+
+func (ldMgr *LogicalDeviceManager) ListLogicalDevicePorts(ctx context.Context, id string) (*voltha.LogicalPorts, error) {
+	log.Debugw("ListLogicalDevicePorts", log.Fields{"logicaldeviceid": id})
+	if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
+		return agent.ListLogicalDevicePorts()
+	}
+	return nil, status.Errorf(codes.NotFound, "%s", id)
+
+}
+
+func (ldMgr *LogicalDeviceManager) getLogicalPort(lPortId *voltha.LogicalPortId) (*voltha.LogicalPort, error) {
+	// Get the logical device where this device is attached
+	var err error
+	var lDevice *voltha.LogicalDevice
+	if lDevice, err = ldMgr.getLogicalDevice(lPortId.Id); err != nil {
+		return nil, err
+	}
+	// Go over list of ports
+	for _, port := range lDevice.Ports {
+		if port.Id == lPortId.PortId {
+			return port, nil
+		}
+	}
+	return nil, status.Errorf(codes.NotFound, "%s-$s", lPortId.Id, lPortId.PortId)
+}
+
+// deleteLogicalPort removes the logical port associated with a child device
+func (ldMgr *LogicalDeviceManager) deleteLogicalPort(ctx context.Context, lPortId *voltha.LogicalPortId) error {
+	log.Debugw("deleting-logical-port", log.Fields{"LDeviceId": lPortId.Id})
+	// Get logical port
+	var logicalPort *voltha.LogicalPort
+	var err error
+	if logicalPort, err = ldMgr.getLogicalPort(lPortId); err != nil {
+		log.Debugw("no-logical-device-port-present", log.Fields{"logicalPortId": lPortId.PortId})
+		return err
+	}
 	// Sanity check
-	if device.Root {
+	if logicalPort.RootPort {
 		return errors.New("Device-root")
 	}
-	logDeviceId, _ := ldMgr.getLogicalDeviceId(device)
-	if logDeviceId == nil {
-		log.Debugw("no-logical-device-present", log.Fields{"deviceId": device.Id})
-		return nil
-	}
-	if agent := ldMgr.getLogicalDeviceAgent(*logDeviceId); agent != nil {
-		agent.deleteLogicalPort(device)
+	if agent := ldMgr.getLogicalDeviceAgent(lPortId.Id); agent != nil {
+		agent.deleteLogicalPort(logicalPort)
 	}
 
 	log.Debug("deleting-logical-port-ends")
@@ -198,7 +258,69 @@
 	log.Debugw("AddUNILogicalPort", log.Fields{"logDeviceId": logDeviceId, "parentId": parentId})
 
 	if agent := ldMgr.getLogicalDeviceAgent(*logDeviceId); agent != nil {
-		return agent.addUNILogicalPort(ctx, childDevice, childDevice.ProxyAddress.ChannelId)
+		return agent.addUNILogicalPort(ctx, childDevice)
 	}
 	return status.Errorf(codes.NotFound, "%s", childDevice.Id)
 }
+
+func (ldMgr *LogicalDeviceManager) updateFlowTable(ctx context.Context, id string, flow *openflow_13.OfpFlowMod, ch chan interface{}) {
+	log.Debugw("updateFlowTable", log.Fields{"logicalDeviceId": id})
+	var res interface{}
+	if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
+		res = agent.updateFlowTable(ctx, flow)
+		log.Debugw("updateFlowTable-result", log.Fields{"result": res})
+	} else {
+		res = status.Errorf(codes.NotFound, "%s", id)
+	}
+	sendAPIResponse(ctx, ch, res)
+}
+
+func (ldMgr *LogicalDeviceManager) updateGroupTable(ctx context.Context, id string, groupMod *openflow_13.OfpGroupMod, ch chan interface{}) {
+	log.Debugw("updateGroupTable", log.Fields{"logicalDeviceId": id})
+	var res interface{}
+	if agent := ldMgr.getLogicalDeviceAgent(id); agent != nil {
+		res = agent.updateGroupTable(ctx, groupMod)
+		log.Debugw("updateGroupTable-result", log.Fields{"result": res})
+	} else {
+		res = status.Errorf(codes.NotFound, "%s", id)
+	}
+	sendAPIResponse(ctx, ch, res)
+}
+
+func (ldMgr *LogicalDeviceManager) enableLogicalPort(ctx context.Context, id *voltha.LogicalPortId, ch chan interface{}) {
+	log.Debugw("enableLogicalPort", log.Fields{"logicalDeviceId": id})
+	var res interface{}
+	// Get logical port
+	var logicalPort *voltha.LogicalPort
+	var err error
+	if logicalPort, err = ldMgr.getLogicalPort(id); err != nil {
+		log.Debugw("no-logical-device-port-present", log.Fields{"logicalPortId": id.PortId})
+		res = err
+	}
+	if agent := ldMgr.getLogicalDeviceAgent(id.Id); agent != nil {
+		res = agent.enableLogicalPort(logicalPort)
+		log.Debugw("enableLogicalPort-result", log.Fields{"result": res})
+	} else {
+		res = status.Errorf(codes.NotFound, "%s", id.Id)
+	}
+	sendAPIResponse(ctx, ch, res)
+}
+
+func (ldMgr *LogicalDeviceManager) disableLogicalPort(ctx context.Context, id *voltha.LogicalPortId, ch chan interface{}) {
+	log.Debugw("disableLogicalPort", log.Fields{"logicalDeviceId": id})
+	var res interface{}
+	// Get logical port
+	var logicalPort *voltha.LogicalPort
+	var err error
+	if logicalPort, err = ldMgr.getLogicalPort(id); err != nil {
+		log.Debugw("no-logical-device-port-present", log.Fields{"logicalPortId": id.PortId})
+		res = err
+	}
+	if agent := ldMgr.getLogicalDeviceAgent(id.Id); agent != nil {
+		res = agent.disableLogicalPort(logicalPort)
+		log.Debugw("disableLogicalPort-result", log.Fields{"result": res})
+	} else {
+		res = status.Errorf(codes.NotFound, "%s", id.Id)
+	}
+	sendAPIResponse(ctx, ch, res)
+}
diff --git a/rw_core/coreIf/device_manager_if.go b/rw_core/coreIf/device_manager_if.go
index c8e33c1..dbe5598 100644
--- a/rw_core/coreIf/device_manager_if.go
+++ b/rw_core/coreIf/device_manager_if.go
@@ -24,4 +24,5 @@
 // DeviceManager represents a generic device manager
 type DeviceManager interface {
 	GetDevice(string) (*voltha.Device, error)
+	IsRootDevice(string) (bool, error)
 }
diff --git a/rw_core/coreIf/logical_device_agent_if.go b/rw_core/coreIf/logical_device_agent_if.go
index b5893ec..43bbf18 100644
--- a/rw_core/coreIf/logical_device_agent_if.go
+++ b/rw_core/coreIf/logical_device_agent_if.go
@@ -27,9 +27,9 @@
 
 // LogicalAgent represents a generic agent
 type LogicalDeviceAgent interface {
-	GetLogicalDevice() *voltha.LogicalDevice
+	GetLogicalDevice() (*voltha.LogicalDevice, error)
 	GetDeviceGraph() *graph.DeviceGraph
 	GetAllDefaultRules() *utils.DeviceRules
 	GetWildcardInputPorts(excludePort ...uint32) []uint32
-	GetRoute(ingressPortNo *uint32, egressPortNo *uint32) []graph.RouteHop
+	GetRoute(ingressPortNo uint32, egressPortNo uint32) []graph.RouteHop
 }
diff --git a/rw_core/flow_decomposition/flow_decomposer.go b/rw_core/flow_decomposition/flow_decomposer.go
index 233465b..284bef2 100644
--- a/rw_core/flow_decomposition/flow_decomposer.go
+++ b/rw_core/flow_decomposition/flow_decomposer.go
@@ -20,6 +20,7 @@
 	"bytes"
 	"crypto/md5"
 	"fmt"
+	"github.com/gogo/protobuf/proto"
 	"github.com/opencord/voltha-go/common/log"
 	ofp "github.com/opencord/voltha-go/protos/openflow_13"
 	"github.com/opencord/voltha-go/protos/voltha"
@@ -175,7 +176,7 @@
 }
 
 func VlanPcp(vlanPcp uint32) *ofp.OfpOxmOfbField {
-	return &ofp.OfpOxmOfbField{Type: VLAN_VID, Value: &ofp.OfpOxmOfbField_VlanPcp{VlanPcp: vlanPcp}}
+	return &ofp.OfpOxmOfbField{Type: VLAN_PCP, Value: &ofp.OfpOxmOfbField_VlanPcp{VlanPcp: vlanPcp}}
 }
 
 func IpDscp(ipDscp uint32) *ofp.OfpOxmOfbField {
@@ -343,6 +344,37 @@
 	return nil
 }
 
+func UpdateOutputPortByActionType(flow *ofp.OfpFlowStats, actionType uint32, toPort uint32) *ofp.OfpFlowStats {
+	if flow == nil {
+		return nil
+	}
+	nFlow := (proto.Clone(flow)).(*ofp.OfpFlowStats)
+	nFlow.Instructions = nil
+	nInsts := make([]*ofp.OfpInstruction, 0)
+	for _, instruction := range flow.Instructions {
+		if instruction.Type == actionType {
+			instActions := instruction.GetActions()
+			if instActions == nil {
+				return nil
+			}
+			nActions := make([]*ofp.OfpAction, 0)
+			for _, action := range instActions.Actions {
+				if action.GetOutput() != nil {
+					nActions = append(nActions, Output(toPort))
+				} else {
+					nActions = append(nActions, action)
+				}
+			}
+			instructionAction := ofp.OfpInstruction_Actions{Actions: &ofp.OfpInstructionActions{Actions: nActions}}
+			nInsts = append(nInsts, &ofp.OfpInstruction{Type: uint32(APPLY_ACTIONS), Data: &instructionAction})
+		} else {
+			nInsts = append(nInsts, instruction)
+		}
+	}
+	nFlow.Instructions = nInsts
+	return nFlow
+}
+
 func excludeOxmOfbField(field *ofp.OfpOxmOfbField, exclude ...ofp.OxmOfbFieldTypes) bool {
 	for _, fieldToExclude := range exclude {
 		if field.Type == fieldToExclude {
@@ -709,6 +741,42 @@
 	return deviceRules
 }
 
+// Handles special case of any controller-bound flow for a parent device
+func (fd *FlowDecomposer) updateOutputPortForControllerBoundFlowForParentDevide(flow *ofp.OfpFlowStats,
+	dr *fu.DeviceRules) *fu.DeviceRules {
+	EAPOL := EthType(0x888e)
+	IGMP := IpProto(2)
+	UDP := IpProto(17)
+
+	newDeviceRules := dr.Copy()
+	//	Check whether we are dealing with a parent device
+	for deviceId, fg := range dr.GetRules() {
+		if root, _ := fd.deviceMgr.IsRootDevice(deviceId); root {
+			newDeviceRules.ClearFlows(deviceId)
+			for i := 0; i < fg.Flows.Len(); i++ {
+				f := fg.GetFlow(i)
+				UpdateOutPortNo := false
+				for _, field := range GetOfbFields(f) {
+					UpdateOutPortNo = (field.String() == EAPOL.String())
+					UpdateOutPortNo = UpdateOutPortNo || (field.String() == IGMP.String())
+					UpdateOutPortNo = UpdateOutPortNo || (field.String() == UDP.String())
+					if UpdateOutPortNo {
+						break
+					}
+				}
+				if UpdateOutPortNo {
+					f = UpdateOutputPortByActionType(f, uint32(ofp.OfpInstructionType_OFPIT_APPLY_ACTIONS),
+						uint32(ofp.OfpPortNo_OFPP_CONTROLLER))
+				}
+				// Update flow Id as a change in the instruction field will result in a new flow ID
+				f.Id = hashFlowStats(f)
+				newDeviceRules.AddFlow(deviceId, (proto.Clone(f)).(*ofp.OfpFlowStats))
+			}
+		}
+	}
+	return newDeviceRules
+}
+
 //processControllerBoundFlow decomposes trap flows
 func (fd *FlowDecomposer) processControllerBoundFlow(agent coreIf.LogicalDeviceAgent, route []graph.RouteHop,
 	inPortNo uint32, outPortNo uint32, flow *ofp.OfpFlowStats) *fu.DeviceRules {
@@ -904,7 +972,7 @@
 		log.Debugw("creating-metadata-flow", log.Fields{"flow": flow})
 		portNumber := uint32(GetPortNumberFromMetadata(flow))
 		if portNumber != 0 {
-			recalculatedRoute := agent.GetRoute(&inPortNo, &portNumber)
+			recalculatedRoute := agent.GetRoute(inPortNo, portNumber)
 			switch len(recalculatedRoute) {
 			case 0:
 				log.Errorw("no-route-double-tag", log.Fields{"inPortNo": inPortNo, "outPortNo": portNumber, "comment": "deleting-flow", "metadata": GetMetaData64Bit(flow)})
@@ -1071,7 +1139,7 @@
 			}
 		}
 
-		route2 := agent.GetRoute(&inPortNo, &outPortNo)
+		route2 := agent.GetRoute(inPortNo, outPortNo)
 		switch len(route2) {
 		case 0:
 			log.Errorw("mc-no-route", log.Fields{"inPortNo": inPortNo, "outPortNo": outPortNo, "comment": "deleting flow"})
@@ -1144,7 +1212,7 @@
 
 	deviceRules := fu.NewDeviceRules()
 
-	route := agent.GetRoute(&inPortNo, &outPortNo)
+	route := agent.GetRoute(inPortNo, outPortNo)
 	switch len(route) {
 	case 0:
 		log.Errorw("no-route", log.Fields{"inPortNo": inPortNo, "outPortNo": outPortNo, "comment": "deleting-flow"})
@@ -1182,5 +1250,6 @@
 			deviceRules = fd.processMulticastFlow(agent, route, inPortNo, outPortNo, flow, grpId, groupMap)
 		}
 	}
+	deviceRules = fd.updateOutputPortForControllerBoundFlowForParentDevide(flow, deviceRules)
 	return deviceRules
 }
diff --git a/rw_core/flow_decomposition/flow_decomposer_test.go b/rw_core/flow_decomposition/flow_decomposer_test.go
index 6c7d7fa..f00a1e7 100644
--- a/rw_core/flow_decomposition/flow_decomposer_test.go
+++ b/rw_core/flow_decomposition/flow_decomposer_test.go
@@ -94,6 +94,12 @@
 	}
 	return nil, errors.New("ABSENT.")
 }
+func (tdm *testDeviceManager) IsRootDevice(deviceId string) (bool, error) {
+	if d, ok := tdm.devices[deviceId]; ok {
+		return d.Root, nil
+	}
+	return false, errors.New("ABSENT.")
+}
 
 type testFlowDecomposer struct {
 	dMgr         *testDeviceManager
@@ -372,8 +378,8 @@
 	return ""
 }
 
-func (tfd *testFlowDecomposer) GetLogicalDevice() *voltha.LogicalDevice {
-	return nil
+func (tfd *testFlowDecomposer) GetLogicalDevice() (*voltha.LogicalDevice, error) {
+	return nil, nil
 }
 
 func (tfd *testFlowDecomposer) GetDeviceGraph() *graph.DeviceGraph {
@@ -398,19 +404,19 @@
 	return lPorts
 }
 
-func (tfd *testFlowDecomposer) GetRoute(ingressPortNo *uint32, egressPortNo *uint32) []graph.RouteHop {
+func (tfd *testFlowDecomposer) GetRoute(ingressPortNo uint32, egressPortNo uint32) []graph.RouteHop {
 	var portLink graph.OFPortLink
-	if egressPortNo == nil {
+	if egressPortNo == 0 {
 		portLink.Egress = 0
-	} else if *egressPortNo&0x7fffffff == uint32(ofp.OfpPortNo_OFPP_CONTROLLER) {
+	} else if egressPortNo&0x7fffffff == uint32(ofp.OfpPortNo_OFPP_CONTROLLER) {
 		portLink.Egress = 10
 	} else {
-		portLink.Egress = *egressPortNo
+		portLink.Egress = egressPortNo
 	}
-	if ingressPortNo == nil {
+	if ingressPortNo == 0 {
 		portLink.Ingress = 0
 	} else {
-		portLink.Ingress = *ingressPortNo
+		portLink.Ingress = ingressPortNo
 	}
 	for key, val := range tfd.routes {
 		if key.Ingress == portLink.Ingress && key.Egress == portLink.Egress {
@@ -472,7 +478,7 @@
 		Actions: []*ofp.OfpAction{
 			PushVlan(0x8100),
 			SetField(VlanVid(uint32(ofp.OfpVlanId_OFPVID_PRESENT) | 4000)),
-			Output(2),
+			Output(uint32(ofp.OfpPortNo_OFPP_CONTROLLER)),
 		},
 	}
 	expectedOltFlow := MkFlowStat(fa)
@@ -556,7 +562,7 @@
 		Actions: []*ofp.OfpAction{
 			PushVlan(0x8100),
 			SetField(VlanVid(uint32(ofp.OfpVlanId_OFPVID_PRESENT) | 4000)),
-			Output(2),
+			Output(uint32(ofp.OfpPortNo_OFPP_CONTROLLER)),
 		},
 	}
 	expectedOltFlow := MkFlowStat(fa)
diff --git a/rw_core/utils/flow_utils.go b/rw_core/utils/flow_utils.go
index 67d7c29..6fa6a02 100644
--- a/rw_core/utils/flow_utils.go
+++ b/rw_core/utils/flow_utils.go
@@ -20,6 +20,7 @@
 	"github.com/cevaris/ordered_map"
 	"github.com/gogo/protobuf/proto"
 	ofp "github.com/opencord/voltha-go/protos/openflow_13"
+	"strings"
 )
 
 type OfpFlowModArgs map[string]uint64
@@ -82,6 +83,28 @@
 	return nil
 }
 
+func (fg *FlowsAndGroups) ListFlows() []*ofp.OfpFlowStats {
+	flows := make([]*ofp.OfpFlowStats, 0)
+	iter := fg.Flows.IterFunc()
+	for kv, ok := iter(); ok; kv, ok = iter() {
+		if protoMsg, isMsg := kv.Value.(*ofp.OfpFlowStats); isMsg {
+			flows = append(flows, protoMsg)
+		}
+	}
+	return flows
+}
+
+func (fg *FlowsAndGroups) ListGroups() []*ofp.OfpGroupEntry {
+	groups := make([]*ofp.OfpGroupEntry, 0)
+	iter := fg.Groups.IterFunc()
+	for kv, ok := iter(); ok; kv, ok = iter() {
+		if protoMsg, isMsg := kv.Value.(*ofp.OfpGroupEntry); isMsg {
+			groups = append(groups, protoMsg)
+		}
+	}
+	return groups
+}
+
 func (fg *FlowsAndGroups) String() string {
 	var buffer bytes.Buffer
 	iter := fg.Flows.IterFunc()
@@ -154,6 +177,23 @@
 	return copyDR
 }
 
+func (dr *DeviceRules) ClearFlows(deviceId string) {
+	if _, exist := dr.Rules[deviceId]; exist {
+		dr.Rules[deviceId].Flows = ordered_map.NewOrderedMap()
+	}
+}
+
+func (dr *DeviceRules) AddFlow(deviceId string, flow *ofp.OfpFlowStats) {
+	if _, exist := dr.Rules[deviceId]; !exist {
+		dr.Rules[deviceId] = NewFlowsAndGroups()
+	}
+	dr.Rules[deviceId].AddFlow(flow)
+}
+
+func (dr *DeviceRules) GetRules() map[string]*FlowsAndGroups {
+	return dr.Rules
+}
+
 func (dr *DeviceRules) String() string {
 	var buffer bytes.Buffer
 	for key, value := range dr.Rules {
@@ -179,3 +219,159 @@
 		dr.Rules[deviceId] = NewFlowsAndGroups()
 	}
 }
+
+/*
+ *  Common flow routines
+ */
+
+//FindOverlappingFlows return a list of overlapping flow(s) where mod is the flow request
+func FindOverlappingFlows(flows []*ofp.OfpFlowStats, mod *ofp.OfpFlowMod) []*ofp.OfpFlowStats {
+	return nil //TODO - complete implementation
+}
+
+// FindFlowById returns the index of the flow in the flows array if present. Otherwise, it returns -1
+func FindFlowById(flows []*ofp.OfpFlowStats, flow *ofp.OfpFlowStats) int {
+	for idx, f := range flows {
+		if flow.Id == f.Id {
+			return idx
+		}
+	}
+	return -1
+}
+
+// FindFlows returns the index in flows where flow if present.  Otherwise, it returns -1
+func FindFlows(flows []*ofp.OfpFlowStats, flow *ofp.OfpFlowStats) int {
+	for idx, f := range flows {
+		if FlowMatch(f, flow) {
+			return idx
+		}
+	}
+	return -1
+}
+
+//FlowMatch returns true if two flows matches on the following flow attributes:
+//TableId, Priority, Flags, Cookie, Match
+func FlowMatch(f1 *ofp.OfpFlowStats, f2 *ofp.OfpFlowStats) bool {
+	keysMatter := []string{"TableId", "Priority", "Flags", "Cookie", "Match"}
+	for _, key := range keysMatter {
+		switch key {
+		case "TableId":
+			if f1.TableId != f2.TableId {
+				return false
+			}
+		case "Priority":
+			if f1.Priority != f2.Priority {
+				return false
+			}
+		case "Flags":
+			if f1.Flags != f2.Flags {
+				return false
+			}
+		case "Cookie":
+			if f1.Cookie != f2.Cookie {
+				return false
+			}
+		case "Match":
+			if strings.Compare(f1.Match.String(), f2.Match.String()) != 0 {
+				return false
+			}
+		}
+	}
+	return true
+}
+
+//FlowMatchesMod returns True if given flow is "covered" by the wildcard flow_mod, taking into consideration of
+//both exact matches as well as masks-based match fields if any. Otherwise return False
+func FlowMatchesMod(flow *ofp.OfpFlowStats, mod *ofp.OfpFlowMod) bool {
+	//Check if flow.cookie is covered by mod.cookie and mod.cookie_mask
+	if (flow.Cookie & mod.CookieMask) != (mod.Cookie & mod.CookieMask) {
+		return false
+	}
+
+	//Check if flow.table_id is covered by flow_mod.table_id
+	if mod.TableId != uint32(ofp.OfpTable_OFPTT_ALL) && flow.TableId != mod.TableId {
+		return false
+	}
+
+	//Check out_port
+	if (mod.OutPort&0x7fffffff) != uint32(ofp.OfpPortNo_OFPP_ANY) && !FlowHasOutPort(flow, mod.OutPort) {
+		return false
+	}
+
+	//	Check out_group
+	if (mod.OutGroup&0x7fffffff) != uint32(ofp.OfpGroup_OFPG_ANY) && !FlowHasOutGroup(flow, mod.OutGroup) {
+		return false
+	}
+
+	//Priority is ignored
+
+	//Check match condition
+	//If the flow_mod match field is empty, that is a special case and indicates the flow entry matches
+	if (mod.Match == nil) || (mod.Match.OxmFields == nil) {
+		//If we got this far and the match is empty in the flow spec, than the flow matches
+		return true
+	} // TODO : implement the flow match analysis
+	return false
+
+}
+
+//FlowHasOutPort returns True if flow has a output command with the given out_port
+func FlowHasOutPort(flow *ofp.OfpFlowStats, outPort uint32) bool {
+	for _, instruction := range flow.Instructions {
+		if instruction.Type == uint32(ofp.OfpInstructionType_OFPIT_APPLY_ACTIONS) {
+			if instruction.GetActions() == nil {
+				return false
+			}
+			for _, action := range instruction.GetActions().Actions {
+				if action.Type == ofp.OfpActionType_OFPAT_OUTPUT {
+					if (action.GetOutput() != nil) && (action.GetOutput().Port == outPort) {
+						return true
+					}
+				}
+
+			}
+		}
+	}
+	return false
+}
+
+//FlowHasOutGroup return True if flow has a output command with the given out_group
+func FlowHasOutGroup(flow *ofp.OfpFlowStats, groupID uint32) bool {
+	for _, instruction := range flow.Instructions {
+		if instruction.Type == uint32(ofp.OfpInstructionType_OFPIT_APPLY_ACTIONS) {
+			if instruction.GetActions() == nil {
+				return false
+			}
+			for _, action := range instruction.GetActions().Actions {
+				if action.Type == ofp.OfpActionType_OFPAT_GROUP {
+					if (action.GetGroup() != nil) && (action.GetGroup().GroupId == groupID) {
+						return true
+					}
+				}
+
+			}
+		}
+	}
+	return false
+}
+
+//FindGroup returns index of group if found, else returns -1
+func FindGroup(groups []*ofp.OfpGroupEntry, groupId uint32) int {
+	for idx, group := range groups {
+		if group.Desc.GroupId == groupId {
+			return idx
+		}
+	}
+	return -1
+}
+
+func FlowsDeleteByGroupId(flows []*ofp.OfpFlowStats, groupId uint32) (bool, []*ofp.OfpFlowStats) {
+	toKeep := make([]*ofp.OfpFlowStats, 0)
+
+	for _, f := range flows {
+		if !FlowHasOutGroup(f, groupId) {
+			toKeep = append(toKeep, f)
+		}
+	}
+	return len(toKeep) < len(flows), toKeep
+}