FPM component that can decode routes from Quagga's FIB push interface.

Change-Id: I57bfd9273b81c8d368a59a3acea53486cb4acfc1

@@ -16,19 +16,19 @@
 package org.onosproject.routing;
 
 /**
- * Provides a way of interacting with the BGP protocol component.
+ * A source of route updates.
  */
-public interface BgpService {
+public interface RouteSourceService {
 
     /**
-     * Starts the BGP service.
+     * Starts the route source.
      *
      * @param routeListener listener to send route updates to
      */
     void start(RouteListener routeListener);
 
     /**
-     * Stops the BGP service.
+     * Stops the route source.
      */
     void stop();
 }

@@ -34,7 +34,7 @@ import org.onlab.packet.Ip4Address;
 import org.onlab.packet.Ip4Prefix;
 import org.onlab.packet.Ip6Prefix;
 import org.onlab.packet.IpPrefix;
-import org.onosproject.routing.BgpService;
+import org.onosproject.routing.RouteSourceService;
 import org.onosproject.routing.RouteListener;
 import org.osgi.service.component.ComponentContext;
 import org.slf4j.Logger;
@@ -57,7 +57,7 @@ import static org.onlab.util.Tools.groupedThreads;
  */
 @Component(immediate = true, enabled = false)
 @Service
-public class BgpSessionManager implements BgpInfoService, BgpService {
+public class BgpSessionManager implements BgpInfoService, RouteSourceService {
     private static final Logger log =
             LoggerFactory.getLogger(BgpSessionManager.class);
 

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.handler.codec.frame.FrameDecoder;
+import org.onosproject.routing.fpm.protocol.FpmHeader;
+
+/**
+ * Frame decoder for FPM connections.
+ */
+public class FpmFrameDecoder extends FrameDecoder {
+
+    @Override
+    protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buffer)
+            throws Exception {
+
+        if (!channel.isConnected()) {
+            return null;
+        }
+
+        if (buffer.readableBytes() < FpmHeader.FPM_HEADER_LENGTH) {
+            return null;
+        }
+
+        buffer.markReaderIndex();
+
+        short version = buffer.readUnsignedByte();
+        short type = buffer.readUnsignedByte();
+        int length = buffer.readUnsignedShort();
+
+        buffer.resetReaderIndex();
+
+        if (buffer.readableBytes() < length) {
+            // Not enough bytes to read a whole message
+            return null;
+        }
+
+        byte[] fpmMessage = new byte[length];
+        buffer.readBytes(fpmMessage);
+
+        return FpmHeader.decode(fpmMessage, 0, fpmMessage.length);
+    }
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.onosproject.routing.fpm;
+
+import org.apache.felix.scr.annotations.Activate;
+import org.apache.felix.scr.annotations.Component;
+import org.apache.felix.scr.annotations.Deactivate;
+import org.apache.felix.scr.annotations.Service;
+import org.jboss.netty.bootstrap.ServerBootstrap;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelException;
+import org.jboss.netty.channel.ChannelFactory;
+import org.jboss.netty.channel.ChannelPipeline;
+import org.jboss.netty.channel.ChannelPipelineFactory;
+import org.jboss.netty.channel.Channels;
+import org.jboss.netty.channel.group.ChannelGroup;
+import org.jboss.netty.channel.group.DefaultChannelGroup;
+import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
+import org.onlab.packet.IpAddress;
+import org.onlab.packet.IpPrefix;
+import org.onosproject.routing.RouteEntry;
+import org.onosproject.routing.RouteListener;
+import org.onosproject.routing.RouteSourceService;
+import org.onosproject.routing.RouteUpdate;
+import org.onosproject.routing.fpm.protocol.FpmHeader;
+import org.onosproject.routing.fpm.protocol.Netlink;
+import org.onosproject.routing.fpm.protocol.RouteAttribute;
+import org.onosproject.routing.fpm.protocol.RouteAttributeDst;
+import org.onosproject.routing.fpm.protocol.RouteAttributeGateway;
+import org.onosproject.routing.fpm.protocol.RtNetlink;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.InetSocketAddress;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static java.util.concurrent.Executors.newCachedThreadPool;
+import static org.onlab.util.Tools.groupedThreads;
+
+/**
+ * Forwarding Plane Manager (FPM) route source.
+ */
+@Service
+@Component(immediate = true, enabled = false)
+public class FpmManager implements RouteSourceService {
+    private final Logger log = LoggerFactory.getLogger(getClass());
+
+    private ServerBootstrap serverBootstrap;
+    private Channel serverChannel;
+    private ChannelGroup allChannels = new DefaultChannelGroup();
+
+    private Map<IpPrefix, RouteEntry> fpmRoutes = new ConcurrentHashMap<>();
+
+    private RouteListener routeListener;
+
+    private static final int FPM_PORT = 2620;
+
+    @Activate
+    protected void activate() {
+        log.info("Started");
+    }
+
+    @Deactivate
+    protected void deactivate() {
+        stopServer();
+        log.info("Stopped");
+    }
+
+    private void startServer() {
+        ChannelFactory channelFactory = new NioServerSocketChannelFactory(
+                newCachedThreadPool(groupedThreads("onos/fpm", "sm-boss-%d")),
+                newCachedThreadPool(groupedThreads("onos/fpm", "sm-worker-%d")));
+        ChannelPipelineFactory pipelineFactory = () -> {
+            // Allocate a new session per connection
+            FpmSessionHandler fpmSessionHandler =
+                    new FpmSessionHandler(new InternalFpmListener());
+            FpmFrameDecoder fpmFrameDecoder =
+                    new FpmFrameDecoder();
+
+            // Setup the processing pipeline
+            ChannelPipeline pipeline = Channels.pipeline();
+            pipeline.addLast("FpmFrameDecoder", fpmFrameDecoder);
+            pipeline.addLast("FpmSession", fpmSessionHandler);
+            return pipeline;
+        };
+
+        InetSocketAddress listenAddress = new InetSocketAddress(FPM_PORT);
+
+        serverBootstrap = new ServerBootstrap(channelFactory);
+        serverBootstrap.setOption("child.reuseAddr", true);
+        serverBootstrap.setOption("child.keepAlive", true);
+        serverBootstrap.setOption("child.tcpNoDelay", true);
+        serverBootstrap.setPipelineFactory(pipelineFactory);
+        try {
+            serverChannel = serverBootstrap.bind(listenAddress);
+            allChannels.add(serverChannel);
+        } catch (ChannelException e) {
+            log.debug("Exception binding to FPM port {}: ",
+                    listenAddress.getPort(), e);
+            stopServer();
+        }
+    }
+
+    private void stopServer() {
+        allChannels.close().awaitUninterruptibly();
+        allChannels.clear();
+        if (serverBootstrap != null) {
+            serverBootstrap.releaseExternalResources();
+        }
+    }
+
+    @Override
+    public void start(RouteListener routeListener) {
+        this.routeListener = routeListener;
+
+        startServer();
+    }
+
+    @Override
+    public void stop() {
+        fpmRoutes.clear();
+        stopServer();
+    }
+
+    private void fpmMessage(FpmHeader fpmMessage) {
+        Netlink netlink = fpmMessage.netlink();
+        RtNetlink rtNetlink = netlink.rtNetlink();
+
+        if (log.isTraceEnabled()) {
+            log.trace("Received FPM message: {}", fpmMessage);
+        }
+
+        IpAddress dstAddress = null;
+        IpAddress gateway = null;
+
+        for (RouteAttribute attribute : rtNetlink.attributes()) {
+            if (attribute.type() == RouteAttribute.RTA_DST) {
+                RouteAttributeDst raDst = (RouteAttributeDst) attribute;
+                dstAddress = raDst.dstAddress();
+            } else if (attribute.type() == RouteAttribute.RTA_GATEWAY) {
+                RouteAttributeGateway raGateway = (RouteAttributeGateway) attribute;
+                gateway = raGateway.gateway();
+            }
+        }
+
+        if (dstAddress == null) {
+            log.error("Dst address missing!");
+            return;
+        }
+
+        IpPrefix prefix = IpPrefix.valueOf(dstAddress, rtNetlink.dstLength());
+
+        RouteUpdate routeUpdate = null;
+        RouteEntry entry;
+        switch (netlink.type()) {
+        case RTM_NEWROUTE:
+            if (gateway == null) {
+                // We ignore interface routes with no gateway for now.
+                return;
+            }
+            entry = new RouteEntry(prefix, gateway);
+
+            fpmRoutes.put(entry.prefix(), entry);
+
+            routeUpdate = new RouteUpdate(RouteUpdate.Type.UPDATE, entry);
+            break;
+        case RTM_DELROUTE:
+            RouteEntry existing = fpmRoutes.remove(prefix);
+            if (existing == null) {
+                log.warn("Got delete for non-existent prefix");
+                return;
+            }
+
+            entry = new RouteEntry(prefix, existing.nextHop());
+
+            routeUpdate = new RouteUpdate(RouteUpdate.Type.DELETE, entry);
+            break;
+        case RTM_GETROUTE:
+        default:
+            break;
+        }
+
+        if (routeUpdate == null) {
+            log.warn("Unsupported FPM message: {}", fpmMessage);
+            return;
+        }
+
+        routeListener.update(Collections.singletonList(routeUpdate));
+    }
+
+    private class InternalFpmListener implements FpmMessageListener {
+        @Override
+        public void fpmMessage(FpmHeader fpmMessage) {
+            FpmManager.this.fpmMessage(fpmMessage);
+        }
+    }
+
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm;
+
+import org.onosproject.routing.fpm.protocol.FpmHeader;
+
+/**
+ * Listener for FPM messages.
+ */
+public interface FpmMessageListener {
+
+    /**
+     * Handles an FPM message.
+     *
+     * @param fpmMessage FPM message
+     */
+    void fpmMessage(FpmHeader fpmMessage);
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm;
+
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.ExceptionEvent;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.SimpleChannelHandler;
+import org.onosproject.routing.fpm.protocol.FpmHeader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+/**
+ * Session handler for FPM protocol.
+ */
+public class FpmSessionHandler extends SimpleChannelHandler {
+
+    private static Logger log = LoggerFactory.getLogger(FpmSessionHandler.class);
+
+    private final FpmMessageListener fpmListener;
+
+    private Channel channel;
+
+    /**
+     * Class constructor.
+     *
+     * @param fpmListener listener for FPM messages
+     */
+    public FpmSessionHandler(FpmMessageListener fpmListener) {
+        this.fpmListener = checkNotNull(fpmListener);
+    }
+
+    @Override
+    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
+            throws Exception {
+        FpmHeader fpmMessage = (FpmHeader) e.getMessage();
+        fpmListener.fpmMessage(fpmMessage);
+    }
+
+    @Override
+    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
+            throws Exception {
+        log.error("Exception thrown while handling FPM message", e.getCause());
+        channel.close();
+        handleDisconnect();
+    }
+
+    @Override
+    public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
+            throws Exception {
+        if (this.channel != null) {
+            log.error("Received new FPM connection while already connected");
+            ctx.getChannel().close();
+            return;
+        }
+
+        this.channel = ctx.getChannel();
+    }
+
+    @Override
+    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)
+            throws Exception {
+        super.channelConnected(ctx, e);
+    }
+
+    @Override
+    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e)
+            throws Exception {
+        handleDisconnect();
+    }
+
+    @Override
+    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e)
+            throws Exception {
+        handleDisconnect();
+    }
+
+    private void handleDisconnect() {
+        this.channel = null;
+    }
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Forwarding Plane Manager (FPM) implementation.
+ */
+package org.onosproject.routing.fpm;

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import com.google.common.base.MoreObjects;
+import org.onlab.packet.DeserializationException;
+
+import java.nio.ByteBuffer;
+
+import static org.onlab.packet.PacketUtils.checkInput;
+
+/**
+ * FPM header.
+ */
+public final class FpmHeader {
+    public static final int FPM_HEADER_LENGTH = 4;
+
+    public static final short FPM_VERSION_1 = 1;
+    public static final short FPM_TYPE_NETLINK = 1;
+
+    private static final String VERSION_NOT_SUPPORTED = "FPM version not supported: ";
+    private static final String TYPE_NOT_SUPPORTED = "FPM type not supported: ";
+
+    private final short version;
+    private final short type;
+    private final int length;
+
+    private final Netlink netlink;
+
+    /**
+     * Class constructor.
+     *
+     * @param version version
+     * @param type type
+     * @param length length
+     * @param netlink netlink header
+     */
+    private FpmHeader(short version, short type, int length, Netlink netlink) {
+        this.version = version;
+        this.type = type;
+        this.length = length;
+        this.netlink = netlink;
+    }
+
+    /**
+     * Returns the protocol version.
+     *
+     * @return protocol version
+     */
+    public short version() {
+        return version;
+    }
+
+    /**
+     * Returns the type.
+     *
+     * @return type
+     */
+    public short type() {
+        return type;
+    }
+
+    /**
+     * Returns the message length.
+     *
+     * @return message length
+     */
+    public int length() {
+        return length;
+    }
+
+    /**
+     * Returns the netlink header.
+     *
+     * @return netlink header
+     */
+    public Netlink netlink() {
+        return netlink;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("version", version)
+                .add("type", type)
+                .add("length", length)
+                .add("netlink", netlink)
+                .toString();
+    }
+
+    /**
+     * Decodes an FPM header from an input buffer.
+     *
+     * @param buffer input buffer
+     * @param start starting position the FPM header
+     * @param length length of the message
+     * @return FPM header
+     * @throws DeserializationException if an FPM header could not be decoded
+     * from the input buffer
+     */
+    public static FpmHeader decode(byte[] buffer, int start, int length) throws
+            DeserializationException {
+        checkInput(buffer, start, length, FPM_HEADER_LENGTH);
+
+        ByteBuffer bb = ByteBuffer.wrap(buffer, start, length);
+
+        short version = bb.get();
+        if (version != FPM_VERSION_1) {
+            throw new DeserializationException(VERSION_NOT_SUPPORTED + version);
+        }
+
+        short type = bb.get();
+        if (type != FPM_TYPE_NETLINK) {
+            throw new DeserializationException(TYPE_NOT_SUPPORTED + type);
+        }
+
+        int messageLength = bb.getShort();
+
+        Netlink netlink = Netlink.decode(buffer, bb.position(), bb.limit() - bb.position());
+
+        return new FpmHeader(version, type, messageLength, netlink);
+    }
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import com.google.common.base.MoreObjects;
+import org.onlab.packet.DeserializationException;
+
+import java.nio.ByteBuffer;
+
+import static org.onlab.packet.PacketUtils.checkInput;
+
+/**
+ * Netlink header.
+ * <p>
+ * Taken from struct nlmsghdr in linux/netlink.h
+ * </p>
+ */
+public final class Netlink {
+
+    public static final int NETLINK_HEADER_LENGTH = 16;
+
+    private final long length;
+    private final NetlinkMessageType type;
+    private final int flags;
+    private final long sequence;
+    private final long processPortId;
+
+    private final RtNetlink rtNetlink;
+
+    /**
+     * Class constructor.
+     *
+     * @param length message length
+     * @param type type
+     * @param flags flags
+     * @param sequence sequence number
+     * @param processPortId port ID
+     * @param rtNetlink netlink routing message
+     */
+    private Netlink(long length, NetlinkMessageType type, int flags, long sequence,
+                    long processPortId, RtNetlink rtNetlink) {
+        this.length = length;
+        this.type = type;
+        this.flags = flags;
+        this.sequence = sequence;
+        this.processPortId = processPortId;
+        this.rtNetlink = rtNetlink;
+    }
+
+    /**
+     * Returns the message length.
+     *
+     * @return length
+     */
+    public long length() {
+        return length;
+    }
+
+    /**
+     * Returns the message type.
+     *
+     * @return message type
+     */
+    public NetlinkMessageType type() {
+        return type;
+    }
+
+    /**
+     * Returns the flags.
+     *
+     * @return flags
+     */
+    public int flags() {
+        return flags;
+    }
+
+    /**
+     * Returns the sequence number.
+     *
+     * @return sequence number
+     */
+    public long sequence() {
+        return sequence;
+    }
+
+    /**
+     * Returns the port ID.
+     *
+     * @return port ID
+     */
+    public long processPortId() {
+        return processPortId;
+    }
+
+    /**
+     * Returns the netlink routing message.
+     *
+     * @return netlink routing message
+     */
+    public RtNetlink rtNetlink() {
+        return rtNetlink;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("length", length)
+                .add("type", type)
+                .add("flags", flags)
+                .add("sequence", sequence)
+                .add("processPortId", processPortId)
+                .add("rtNetlink", rtNetlink)
+                .toString();
+    }
+
+    /**
+     * Decodes a netlink header from an input buffer.
+     *
+     * @param buffer input buffer
+     * @param start starting position the netlink header
+     * @param length length of the message
+     * @return netlink header
+     * @throws DeserializationException if a netlink header could not be
+     * decoded from the input buffer
+     */
+    public static Netlink decode(byte[] buffer, int start, int length) throws
+            DeserializationException {
+        checkInput(buffer, start, length, NETLINK_HEADER_LENGTH);
+
+        ByteBuffer bb = ByteBuffer.wrap(buffer, start, length);
+
+        long messageLength = Integer.reverseBytes(bb.getInt());
+        int type = Short.reverseBytes(bb.getShort());
+        int flags = Short.reverseBytes(bb.getShort());
+        long sequence = Integer.reverseBytes(bb.getInt());
+        long processPortId = Integer.reverseBytes(bb.getInt());
+
+        NetlinkMessageType messageType = NetlinkMessageType.get(type);
+        if (messageType == null) {
+            throw new DeserializationException(
+                    "Unsupported Netlink message type: " + type);
+        }
+
+        // Netlink messages from Quagga's FPM protocol are always in the
+        // netlink route family (family 0).
+        RtNetlink rtNetlink = RtNetlink.decode(buffer, bb.position(),
+                bb.limit() - bb.position());
+
+        return new Netlink(messageLength,
+                messageType,
+                flags,
+                sequence,
+                processPortId,
+                rtNetlink);
+    }
+
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+/**
+ * Netlink message types.
+ * <p>
+ * This is a subset of the types used for routing messages (rtnelink).
+ * Taken from linux/rtnetlink.h
+ * </p>
+ */
+public enum NetlinkMessageType {
+    RTM_NEWROUTE(24),
+    RTM_DELROUTE(25),
+    RTM_GETROUTE(26);
+
+    private final int type;
+
+    /**
+     * Enum constructor.
+     *
+     * @param type integer type value
+     */
+    NetlinkMessageType(int type) {
+        this.type = type;
+    }
+
+    /**
+     * Returns the integer type value for this message type.
+     *
+     * @return type value
+     */
+    public int type() {
+        return type;
+    }
+
+    /**
+     * Gets the NetlinkMessageType for the given integer type value.
+     *
+     * @param type type value
+     * @return Netlink message type, or null if unsupported type value
+     */
+    public static NetlinkMessageType get(int type) {
+        for (NetlinkMessageType m : NetlinkMessageType.values()) {
+            if (m.type() == type) {
+                return m;
+            }
+        }
+        return null;
+    }
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import com.google.common.collect.ImmutableMap;
+import org.onlab.packet.DeserializationException;
+
+import java.nio.ByteBuffer;
+import java.util.Map;
+
+import static org.onlab.packet.PacketUtils.checkInput;
+
+/**
+ * Route attribute header.
+ */
+public abstract class RouteAttribute {
+
+    public static final int ROUTE_ATTRIBUTE_HEADER_LENGTH = 4;
+
+    public static final int RTA_DST = 1;
+    public static final int RTA_OIF = 4;
+    public static final int RTA_GATEWAY = 5;
+    public static final int RTA_PRIORITY = 6;
+
+    private final int length;
+    private final int type;
+
+    private static final Map<Integer, RouteAttributeDecoder<?>> TYPE_DECODER_MAP
+            = ImmutableMap.<Integer, RouteAttributeDecoder<?>>builder()
+            .put(RTA_DST, RouteAttributeDst.decoder())
+            .put(RTA_OIF, RouteAttributeOif.decoder())
+            .put(RTA_GATEWAY, RouteAttributeGateway.decoder())
+            .put(RTA_PRIORITY, RouteAttributePriority.decoder())
+            .build();
+
+    /**
+     * Class constructor.
+     *
+     * @param length attribute length
+     * @param type attribute type
+     */
+    protected RouteAttribute(int length, int type) {
+        this.length = length;
+        this.type = type;
+    }
+
+    /**
+     * Returns the attribute length.
+     *
+     * @return length
+     */
+    public int length() {
+        return length;
+    }
+
+    /**
+     * Returns the attribute type.
+     *
+     * @return type
+     */
+    public int type() {
+        return type;
+    }
+
+    @Override
+    public abstract String toString();
+
+    /**
+     * Decodes a route attribute from an input buffer.
+     *
+     * @param buffer input buffer
+     * @param start starting position the route attribute message
+     * @param length length of the message
+     * @return route attribute message
+     * @throws DeserializationException if a route attribute could not be
+     * decoded from the input buffer
+     */
+    public static RouteAttribute decode(byte[] buffer, int start, int length)
+            throws DeserializationException {
+        checkInput(buffer, start, length, ROUTE_ATTRIBUTE_HEADER_LENGTH);
+
+        ByteBuffer bb = ByteBuffer.wrap(buffer, start, length);
+
+        int tlvLength = Short.reverseBytes(bb.getShort());
+        int type = Short.reverseBytes(bb.getShort());
+
+        if (bb.remaining() < tlvLength - ROUTE_ATTRIBUTE_HEADER_LENGTH) {
+            throw new DeserializationException(
+                    "Incorrect buffer size when decoding route attribute");
+        }
+
+        byte[] value = new byte[tlvLength - ROUTE_ATTRIBUTE_HEADER_LENGTH];
+        bb.get(value);
+
+        RouteAttributeDecoder<?> decoder = TYPE_DECODER_MAP.get(type);
+        if (decoder == null) {
+            throw new DeserializationException(
+                    "No decoder found for route attribute type " + type);
+        }
+
+        return decoder.decodeAttribute(tlvLength, type, value);
+    }
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import org.onlab.packet.DeserializationException;
+
+/**
+ * Decoder for a route attribute.
+ */
+@FunctionalInterface
+public interface RouteAttributeDecoder<A extends RouteAttribute> {
+
+    /**
+     * Decodes the a route attribute from the input buffer.
+     *
+     * @param length length of the attribute
+     * @param type type of the attribute
+     * @param value input buffer
+     * @return route attribute
+     * @throws DeserializationException if a route attribute could not be
+     * decoded from the input buffer
+     */
+    A decodeAttribute(int length, int type, byte[] value)
+            throws DeserializationException;
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import com.google.common.base.MoreObjects;
+import org.onlab.packet.DeserializationException;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.Ip6Address;
+import org.onlab.packet.IpAddress;
+
+/**
+ * Destination address route attribute.
+ */
+public final class RouteAttributeDst extends RouteAttribute {
+
+    private final IpAddress dstAddress;
+
+    /**
+     * Class constructor.
+     *
+     * @param length length
+     * @param type type
+     * @param dstAddress destination address
+     */
+    private RouteAttributeDst(int length, int type, IpAddress dstAddress) {
+        super(length, type);
+
+        this.dstAddress = dstAddress;
+    }
+
+    /**
+     * Returns the destination IP address.
+     *
+     * @return destination IP address
+     */
+    public IpAddress dstAddress() {
+        return dstAddress;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("type", type())
+                .add("length", length())
+                .add("dstAddress", dstAddress)
+                .toString();
+    }
+
+    /**
+     * Returns a decoder for a destination address route attribute.
+     *
+     * @return destination address route attribute decoder
+     */
+    public static RouteAttributeDecoder<RouteAttributeDst> decoder() {
+        return (int length, int type, byte[] value) -> {
+
+            IpAddress dstAddress;
+            if (value.length == Ip4Address.BYTE_LENGTH) {
+                dstAddress = IpAddress.valueOf(IpAddress.Version.INET, value);
+            } else if (value.length == Ip6Address.BYTE_LENGTH) {
+                dstAddress = IpAddress.valueOf(IpAddress.Version.INET6, value);
+            } else {
+                throw new DeserializationException("Invalid address length");
+            }
+
+            return new RouteAttributeDst(length, type, dstAddress);
+        };
+    }
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import com.google.common.base.MoreObjects;
+import org.onlab.packet.DeserializationException;
+import org.onlab.packet.Ip4Address;
+import org.onlab.packet.Ip6Address;
+import org.onlab.packet.IpAddress;
+
+/**
+ * Gateway route attribute.
+ */
+public final class RouteAttributeGateway extends RouteAttribute {
+
+    public static final int VALUE_LENGTH = 4;
+
+    private final IpAddress gateway;
+
+    /**
+     * Class constructor.
+     *
+     * @param length length
+     * @param type type
+     * @param gateway gateway address
+     */
+    private RouteAttributeGateway(int length, int type, IpAddress gateway) {
+        super(length, type);
+
+        this.gateway = gateway;
+    }
+
+    /**
+     * Returns the gateway address.
+     *
+     * @return gateway address
+     */
+    public IpAddress gateway() {
+        return gateway;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("type", type())
+                .add("length", length())
+                .add("gateway", gateway)
+                .toString();
+    }
+
+    /**
+     * Returns a decoder for a gateway route attribute.
+     *
+     * @return gateway route attribute decoder
+     */
+    public static RouteAttributeDecoder<RouteAttributeGateway> decoder() {
+        return (int length, int type, byte[] value) -> {
+
+            IpAddress gateway;
+            if (value.length == Ip4Address.BYTE_LENGTH) {
+                gateway = IpAddress.valueOf(IpAddress.Version.INET, value);
+            } else if (value.length == Ip6Address.BYTE_LENGTH) {
+                gateway = IpAddress.valueOf(IpAddress.Version.INET6, value);
+            } else {
+                throw new DeserializationException("Invalid address length");
+            }
+
+            return new RouteAttributeGateway(length, type, gateway);
+        };
+    }
+
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import com.google.common.base.MoreObjects;
+import org.onlab.packet.DeserializationException;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Output interface route attribute.
+ */
+public final class RouteAttributeOif extends RouteAttribute {
+
+    private static final int VALUE_LENGTH = 4;
+
+    private final long outputInterface;
+
+    /**
+     * Class constructor.
+     *
+     * @param length length
+     * @param type type
+     * @param outputInterface output interface
+     */
+    private RouteAttributeOif(int length, int type, long outputInterface) {
+        super(length, type);
+
+        this.outputInterface = outputInterface;
+    }
+
+    /**
+     * Returns the output interface.
+     *
+     * @return output interface
+     */
+    public long outputInterface() {
+        return outputInterface;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("type", type())
+                .add("length", length())
+                .add("outputInterface", outputInterface)
+                .toString();
+    }
+
+    /**
+     * Returns a decoder for a output interface route attribute.
+     *
+     * @return output interface route attribute decoder
+     */
+    public static RouteAttributeDecoder<RouteAttributeOif> decoder() {
+        return (int length, int type, byte[] value) -> {
+            if (value.length != VALUE_LENGTH) {
+                throw new DeserializationException("Wrong value length");
+            }
+
+            long outputInterface = Integer.reverseBytes(ByteBuffer.wrap(value).getInt());
+
+            return new RouteAttributeOif(length, type, outputInterface);
+        };
+    }
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import com.google.common.base.MoreObjects;
+import org.onlab.packet.DeserializationException;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Priority route attribute.
+ */
+public final class RouteAttributePriority extends RouteAttribute {
+
+    private static final int VALUE_LENGTH = 4;
+
+    private final long priority;
+
+    /**
+     * Class constructor.
+     *
+     * @param length length
+     * @param type type
+     * @param priority priority
+     */
+    private RouteAttributePriority(int length, int type, long priority) {
+        super(length, type);
+
+        this.priority = priority;
+    }
+
+    /**
+     * Returns the priority.
+     *
+     * @return priority
+     */
+    public long priority() {
+        return priority;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("type", type())
+                .add("length", length())
+                .add("priority", priority)
+                .toString();
+    }
+
+    /**
+     * Returns a decoder for a priority route attribute.
+     *
+     * @return priority route attribute decoder
+     */
+    public static RouteAttributeDecoder<RouteAttributePriority> decoder() {
+        return (int length, int type, byte[] value) -> {
+            if (value.length != VALUE_LENGTH) {
+                throw new DeserializationException("Wrong value length");
+            }
+
+            long priority = Integer.reverseBytes(ByteBuffer.wrap(value).getInt());
+
+            return new RouteAttributePriority(length, type, priority);
+        };
+    }
+
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.onosproject.routing.fpm.protocol;
+
+import com.google.common.base.MoreObjects;
+import com.google.common.collect.ImmutableList;
+import org.onlab.packet.DeserializationException;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.onlab.packet.PacketUtils.checkInput;
+
+/**
+ * Netlink routing message (rtnetlink).
+ * <p>
+ * Taken from struct rtmsg in linux/rtnetlink.h
+ * </p>
+ */
+public final class RtNetlink {
+
+    private static final int RT_NETLINK_LENGTH = 12;
+
+    private static final int MASK = 0xff;
+
+    private final short addressFamily;
+    private final int dstLength;
+    private final int srcLength;
+    private final short tos;
+    private final short table;
+    private final short protocol;
+    private final short scope;
+    private final short type;
+    private final long flags;
+
+    private final List<RouteAttribute> attributes;
+
+    /**
+     * Class constructor.
+     *
+     * @param addressFamily address family
+     * @param dstLength destination address length
+     * @param srcLength source address length
+     * @param tos type of service
+     * @param table routing table
+     * @param protocol protocol
+     * @param scope scope
+     * @param type type
+     * @param flags flags
+     * @param attributes list of attributes
+     */
+    private RtNetlink(short addressFamily,
+                      int dstLength,
+                      int srcLength,
+                      short tos,
+                      short table,
+                      short protocol,
+                      short scope,
+                      short type,
+                      long flags,
+                      List<RouteAttribute> attributes) {
+
+        this.addressFamily = addressFamily;
+        this.dstLength = dstLength;
+        this.srcLength = srcLength;
+        this.tos = tos;
+        this.table = table;
+        this.protocol = protocol;
+        this.scope = scope;
+        this.type = type;
+        this.flags = flags;
+
+        this.attributes = ImmutableList.copyOf(attributes);
+
+    }
+
+    /**
+     * Returns the address family of the route.
+     *
+     * @return address family
+     */
+    public short addressFamily() {
+        return addressFamily;
+    }
+
+    /**
+     * Returns the destination address length.
+     *
+     * @return destination address length
+     */
+    public int dstLength() {
+        return dstLength;
+    }
+
+    /**
+     * Returns the source address length.
+     *
+     * @return source address length
+     */
+    public int srcLength() {
+        return srcLength;
+    }
+
+    /**
+     * Returns the type of service.
+     *
+     * @return type of service
+     */
+    public short tos() {
+        return tos;
+    }
+
+    /**
+     * Returns the routing table.
+     *
+     * @return routing table
+     */
+    public short table() {
+        return table;
+    }
+
+    /**
+     * Returns the protocol.
+     *
+     * @return protocol
+     */
+    public short protocol() {
+        return protocol;
+    }
+
+    /**
+     * Returns the route scope.
+     *
+     * @return scope
+     */
+    public short scope() {
+        return scope;
+    }
+
+    /**
+     * Returns the route type.
+     *
+     * @return route type
+     */
+    public short type() {
+        return type;
+    }
+
+    /**
+     * Returns the route flags.
+     *
+     * @return route flags
+     */
+    public long flags() {
+        return flags;
+    }
+
+    /**
+     * Returns the list of route attributes.
+     *
+     * @return route attributes
+     */
+    public List<RouteAttribute> attributes() {
+        return attributes;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(getClass())
+                .add("addressFamily", addressFamily)
+                .add("dstLength", dstLength)
+                .add("srcLength", srcLength)
+                .add("tos", tos)
+                .add("table", table)
+                .add("protocol", protocol)
+                .add("scope", scope)
+                .add("type", type)
+                .add("flags", flags)
+                .add("attributes", attributes)
+                .toString();
+    }
+
+    /**
+     * Decodes an rtnetlink message from an input buffer.
+     *
+     * @param buffer input buffer
+     * @param start starting position the rtnetlink message
+     * @param length length of the message
+     * @return rtnetlink message
+     * @throws DeserializationException if an rtnetlink message could not be
+     * decoded from the input buffer
+     */
+    public static RtNetlink decode(byte[] buffer, int start, int length)
+            throws DeserializationException {
+        checkInput(buffer, start, length, RT_NETLINK_LENGTH);
+
+        ByteBuffer bb = ByteBuffer.wrap(buffer, start, length);
+
+        short addressFamily = (short) (bb.get() & MASK);
+        int dstLength = bb.get() & MASK;
+        int srcLength = bb.get() & MASK;
+        short tos = (short) (bb.get() & MASK);
+        short table = (short) (bb.get() & MASK);
+        short protocol = (short) (bb.get() & MASK);
+        short scope = (short) (bb.get() & MASK);
+        short type = (short) (bb.get() & MASK);
+        long flags = Integer.reverseBytes(bb.getInt());
+        List<RouteAttribute> attributes = new ArrayList<>();
+
+        while (bb.hasRemaining()) {
+            RouteAttribute attribute = RouteAttribute.decode(buffer, bb.position(),
+                    bb.limit() - bb.position());
+            attributes.add(attribute);
+            bb.position(bb.position() + attribute.length());
+        }
+
+        return new RtNetlink(
+                addressFamily,
+                dstLength,
+                srcLength,
+                tos,
+                table,
+                protocol,
+                scope,
+                type,
+                flags,
+                attributes);
+    }
+
+}

+/*
+ * Copyright 2016 Open Networking Laboratory
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * FPM protocol implementation.
+ */
+package org.onosproject.routing.fpm.protocol;

@@ -39,7 +39,7 @@ import org.onosproject.net.Host;
 import org.onosproject.net.host.HostEvent;
 import org.onosproject.net.host.HostListener;
 import org.onosproject.net.host.HostService;
-import org.onosproject.routing.BgpService;
+import org.onosproject.routing.RouteSourceService;
 import org.onosproject.routing.FibEntry;
 import org.onosproject.routing.FibListener;
 import org.onosproject.routing.FibUpdate;
@@ -103,7 +103,7 @@ public class Router implements RoutingService {
     protected HostService hostService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
-    protected BgpService bgpService;
+    protected RouteSourceService routeSourceService;
 
     @Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
     protected RoutingConfigurationService routingConfigurationService;
@@ -142,14 +142,14 @@ public class Router implements RoutingService {
     public void start() {
         this.hostService.addListener(hostListener);
 
-        bgpService.start(new InternalRouteListener());
+        routeSourceService.start(new InternalRouteListener());
 
         bgpUpdatesExecutor.execute(this::doUpdatesThread);
     }
 
     @Override
     public void stop() {
-        bgpService.stop();
+        routeSourceService.stop();
 
         this.hostService.removeListener(hostListener);
 

@@ -40,7 +40,7 @@ import org.onosproject.net.host.HostService;
 import org.onosproject.net.provider.ProviderId;
 import org.onosproject.routing.config.RoutingConfigurationService;
 import org.onosproject.routing.impl.Router.InternalHostListener;
-import org.onosproject.routing.BgpService;
+import org.onosproject.routing.RouteSourceService;
 import org.onosproject.routing.FibEntry;
 import org.onosproject.routing.FibListener;
 import org.onosproject.routing.FibUpdate;
@@ -83,10 +83,10 @@ public class RouterAsyncArpTest {
         routingConfigurationService =
                 createMock(RoutingConfigurationService.class);
 
-        BgpService bgpService = createMock(BgpService.class);
-        bgpService.start(anyObject(RouteListener.class));
-        bgpService.stop();
-        replay(bgpService);
+        RouteSourceService routeSourceService = createMock(RouteSourceService.class);
+        routeSourceService.start(anyObject(RouteListener.class));
+        routeSourceService.stop();
+        replay(routeSourceService);
 
         fibListener = createMock(FibListener.class);
 
@@ -94,7 +94,7 @@ public class RouterAsyncArpTest {
         router.coreService = createNiceMock(CoreService.class);
         router.hostService = hostService;
         router.routingConfigurationService = routingConfigurationService;
-        router.bgpService = bgpService;
+        router.routeSourceService = routeSourceService;
         router.activate();
 
         router.addFibListener(fibListener);

@@ -39,7 +39,7 @@ import org.onosproject.net.PortNumber;
 import org.onosproject.net.host.HostListener;
 import org.onosproject.net.host.HostService;
 import org.onosproject.net.provider.ProviderId;
-import org.onosproject.routing.BgpService;
+import org.onosproject.routing.RouteSourceService;
 import org.onosproject.routing.FibEntry;
 import org.onosproject.routing.FibListener;
 import org.onosproject.routing.FibUpdate;
@@ -98,10 +98,10 @@ public class RouterTest {
         routingConfigurationService =
                 createMock(RoutingConfigurationService.class);
 
-        BgpService bgpService = createMock(BgpService.class);
-        bgpService.start(anyObject(RouteListener.class));
-        bgpService.stop();
-        replay(bgpService);
+        RouteSourceService routeSourceService = createMock(RouteSourceService.class);
+        routeSourceService.start(anyObject(RouteListener.class));
+        routeSourceService.stop();
+        replay(routeSourceService);
 
         fibListener = createMock(FibListener.class);
 
@@ -109,7 +109,7 @@ public class RouterTest {
         router.coreService = createNiceMock(CoreService.class);
         router.hostService = hostService;
         router.routingConfigurationService = routingConfigurationService;
-        router.bgpService = bgpService;
+        router.routeSourceService = routeSourceService;
         router.activate();
 
         router.addFibListener(fibListener);

@@ -384,8 +384,11 @@ public class SoftRouterPipeline extends AbstractHandlerBehaviour implements Pipe
                 .fromApp(fwd.appId())
                 .withPriority(fwd.priority())
                 .forDevice(deviceId)
-                .withSelector(filteredSelector)
-                .withTreatment(tt);
+                .withSelector(filteredSelector);
+
+        if (tt != null) {
+            ruleBuilder.withTreatment(tt);
+        }
 
         if (fwd.permanent()) {
             ruleBuilder.makePermanent();