BGP Controller test

Change-Id: I4bdafeec877d5fbfa79306717cf3033936e0fe59

@@ -31,7 +31,7 @@ import com.google.common.base.MoreObjects;
 /**
 /**
  * Provides BGP keep alive message.
  * Provides BGP keep alive message.
  */
  */
-class BGPKeepaliveMsgVer4 implements BGPKeepaliveMsg {
+public class BGPKeepaliveMsgVer4 implements BGPKeepaliveMsg {
 
 
     /*
     /*
     <Keepalive Message>::= <Common Header>
     <Keepalive Message>::= <Common Header>
@@ -88,7 +88,7 @@ class BGPKeepaliveMsgVer4 implements BGPKeepaliveMsg {
     /**
     /**
      * Default constructor.
      * Default constructor.
      */
      */
-    BGPKeepaliveMsgVer4() {
+    public BGPKeepaliveMsgVer4() {
     }
     }
 
 
     /**
     /**

@@ -204,6 +204,15 @@ public class BGPControllerImpl implements BGPController {
 
 
     }
     }
 
 
+    /**
+     * Returns controller.
+     *
+     * @return controller
+     */
+    public Controller controller() {
+        return this.ctrl;
+    }
+
     @Override
     @Override
     public ConcurrentHashMap<BGPId, BGPPeer> connectedPeers() {
     public ConcurrentHashMap<BGPId, BGPPeer> connectedPeers() {
         return connectedPeers;
         return connectedPeers;

@@ -26,6 +26,7 @@ import java.util.concurrent.Executors;
 
 
 import org.jboss.netty.bootstrap.ClientBootstrap;
 import org.jboss.netty.bootstrap.ClientBootstrap;
 import org.jboss.netty.bootstrap.ServerBootstrap;
 import org.jboss.netty.bootstrap.ServerBootstrap;
+import org.jboss.netty.channel.Channel;
 import org.jboss.netty.channel.ChannelPipelineFactory;
 import org.jboss.netty.channel.ChannelPipelineFactory;
 import org.jboss.netty.channel.group.ChannelGroup;
 import org.jboss.netty.channel.group.ChannelGroup;
 import org.jboss.netty.channel.group.DefaultChannelGroup;
 import org.jboss.netty.channel.group.DefaultChannelGroup;
@@ -49,9 +50,12 @@ public class Controller {
     private static final BGPFactory FACTORY4 = BGPFactories.getFactory(BGPVersion.BGP_4);
     private static final BGPFactory FACTORY4 = BGPFactories.getFactory(BGPVersion.BGP_4);
 
 
     private ChannelGroup cg;
     private ChannelGroup cg;
+    public Channel serverChannel;
 
 
     // Configuration options
     // Configuration options
     private static final short BGP_PORT_NUM = 179;
     private static final short BGP_PORT_NUM = 179;
+    private static final short PORT_NUM_ZERO = 0;
+    private static boolean isPortNumSet = false;
     private final int workerThreads = 16;
     private final int workerThreads = 16;
     private final int peerWorkerThreads = 16;
     private final int peerWorkerThreads = 16;
 
 
@@ -119,7 +123,8 @@ public class Controller {
             bootstrap.setPipelineFactory(pfact);
             bootstrap.setPipelineFactory(pfact);
             InetSocketAddress sa = new InetSocketAddress(getBgpPortNum());
             InetSocketAddress sa = new InetSocketAddress(getBgpPortNum());
             cg = new DefaultChannelGroup();
             cg = new DefaultChannelGroup();
-            cg.add(bootstrap.bind(sa));
+            serverChannel = bootstrap.bind(sa);
+            cg.add(serverChannel);
             log.info("Listening for Peer connection on {}", sa);
             log.info("Listening for Peer connection on {}", sa);
         } catch (Exception e) {
         } catch (Exception e) {
             throw new RuntimeException(e);
             throw new RuntimeException(e);
@@ -234,6 +239,16 @@ public class Controller {
      * @return port number
      * @return port number
      */
      */
     public static short getBgpPortNum() {
     public static short getBgpPortNum() {
+        if (isPortNumSet) {
+            return PORT_NUM_ZERO;
+        }
         return BGP_PORT_NUM;
         return BGP_PORT_NUM;
     }
     }
+
+    /**
+     * sets the isPortNumSet as true.
+     */
+    public void setBgpPortNum() {
+        isPortNumSet = true;
+    }
 }
 }
\ No newline at end of file

+/*
+ * Copyright 2014-2015 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.bgp;
+
+import com.google.common.net.InetAddresses;
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.channel.Channel;
+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.socket.nio.NioClientSocketChannelFactory;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.hamcrest.core.Is.is;
+import static org.junit.Assert.assertThat;
+
+import org.onlab.junit.TestUtils;
+
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.SocketAddress;
+import java.util.LinkedList;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.onosproject.bgp.controller.BGPCfg;
+import org.onosproject.bgp.controller.impl.BGPControllerImpl;
+import org.onosproject.bgpio.types.BGPValueType;
+import org.onosproject.bgpio.types.FourOctetAsNumCapabilityTlv;
+import org.onosproject.bgpio.types.MultiProtocolExtnCapabilityTlv;
+
+/**
+ * Test case for BGPControllerImpl.
+ */
+public class BGPControllerImplTest {
+
+    protected static final Logger log = LoggerFactory
+            .getLogger(BGPControllerImplTest.class);
+
+    private static final String IP_LOOPBACK_ID1 = "127.0.0.1";
+
+    private static final int MESSAGE_TIMEOUT_MS = 3000;
+    public byte version;
+    public short asNumber;
+    public short holdTime;
+    public int bgpId = InetAddresses.coerceToInteger(InetAddresses.forString(IP_LOOPBACK_ID1));
+    public boolean isLargeAsCapabilitySet = false;
+    public LinkedList<BGPValueType> capabilityTlv = new LinkedList<>();
+
+    @Before
+    public void setUp() throws Exception {
+        peer1 = new BgpPeerTest(version, asNumber,
+                holdTime, bgpId, isLargeAsCapabilitySet,
+                capabilityTlv);
+
+        bgpControllerImpl = new BGPControllerImpl();
+
+        // NOTE: We use port 0 to bind on any available port
+        bgpControllerImpl.controller().setBgpPortNum();
+        bgpControllerImpl.activate();
+
+        Channel serverChannel = TestUtils.getField(bgpControllerImpl.controller(),
+                                                  "serverChannel");
+        SocketAddress socketAddress = serverChannel.getLocalAddress();
+        InetSocketAddress inetSocketAddress =
+           (InetSocketAddress) socketAddress;
+        InetAddress connectToAddress = InetAddresses.forString("127.0.0.1");
+        connectToSocket = new InetSocketAddress(connectToAddress,
+                       inetSocketAddress.getPort());
+
+        bgpControllerImpl.getConfig().setRouterId("1.1.1.1");
+        bgpControllerImpl.getConfig().setAsNumber(200);
+        bgpControllerImpl.getConfig().setHoldTime((short) 120);
+        bgpControllerImpl.getConfig().setState(BGPCfg.State.IP_AS_CONFIGURED);
+
+        bgpControllerImpl.getConfig().addPeer("127.0.0.1", 200);
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        bgpControllerImpl.deactivate();
+        bgpControllerImpl = null;
+    }
+
+    private BGPControllerImpl bgpControllerImpl;
+
+    BgpPeerTest peer1;
+
+    // The socket that the remote peers should connect to
+    private InetSocketAddress connectToSocket;
+
+    @Test
+    public void bgpOpenMessageTest1() throws InterruptedException {
+        peer1.peerChannelHandler.asNumber = 200;
+        peer1.peerChannelHandler.version = 4;
+        peer1.peerChannelHandler.holdTime = 120;
+        peer1.connect(connectToSocket);
+        boolean result;
+        result = peer1.peerFrameDecoder.receivedOpenMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+        result = peer1.peerFrameDecoder.receivedKeepaliveMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+    }
+
+    @Test
+    public void bgpOpenMessageTest2() throws InterruptedException {
+        // Open message with as number which is not configured at peer
+        peer1.peerChannelHandler.asNumber = 500;
+        peer1.peerChannelHandler.version = 4;
+        peer1.peerChannelHandler.holdTime = 120;
+        peer1.connect(connectToSocket);
+
+        boolean result;
+        result = peer1.peerFrameDecoder.receivedNotificationMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+    }
+
+    @Test
+    public void bgpOpenMessageTest3() throws InterruptedException {
+        // Open message with invalid hold time value
+        peer1.peerChannelHandler.asNumber = 200;
+        peer1.peerChannelHandler.version = 4;
+        peer1.peerChannelHandler.holdTime = 1;
+        peer1.connect(connectToSocket);
+
+        boolean result;
+        result = peer1.peerFrameDecoder.receivedNotificationMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+    }
+
+    @Test
+    public void bgpOpenMessageTest4() throws InterruptedException {
+        // Open message with invalid as number
+        peer1.peerChannelHandler.asNumber = 200;
+        peer1.peerChannelHandler.version = 4;
+        peer1.peerChannelHandler.holdTime = 120;
+        peer1.peerChannelHandler.isLargeAsCapabilitySet = true;
+        BGPValueType tempTlv = new FourOctetAsNumCapabilityTlv(766545);
+        peer1.peerChannelHandler.capabilityTlv.add(tempTlv);
+        peer1.connect(connectToSocket);
+
+        boolean result;
+        result = peer1.peerFrameDecoder.receivedNotificationMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+    }
+
+    @Test
+    public void bgpOpenMessageTest5() throws InterruptedException {
+        // Open message with LS capability
+        short afi = 16388;
+        byte res = 0;
+        byte safi = 71;
+        peer1.peerChannelHandler.asNumber = 200;
+        peer1.peerChannelHandler.version = 4;
+        peer1.peerChannelHandler.holdTime = 120;
+        bgpControllerImpl.getConfig().setLsCapability(true);
+        BGPValueType tempTlv1 = new MultiProtocolExtnCapabilityTlv(afi, res, safi);
+        peer1.peerChannelHandler.capabilityTlv.add(tempTlv1);
+        peer1.connect(connectToSocket);
+
+        boolean result;
+        result = peer1.peerFrameDecoder.receivedOpenMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+        result = peer1.peerFrameDecoder.receivedKeepaliveMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+    }
+
+    @Test
+    public void bgpOpenMessageTest6() throws InterruptedException {
+        // Open message with as4 capability
+        peer1.peerChannelHandler.asNumber = 200;
+        peer1.peerChannelHandler.version = 4;
+        peer1.peerChannelHandler.holdTime = 120;
+        peer1.peerChannelHandler.isLargeAsCapabilitySet = true;
+        bgpControllerImpl.getConfig().setLargeASCapability(true);
+        BGPValueType tempTlv = new FourOctetAsNumCapabilityTlv(200);
+        peer1.peerChannelHandler.capabilityTlv.add(tempTlv);
+        peer1.connect(connectToSocket);
+
+        boolean result;
+        result = peer1.peerFrameDecoder.receivedOpenMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+        result = peer1.peerFrameDecoder.receivedKeepaliveMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+
+        result = peer1.peerFrameDecoder.receivedKeepaliveMessageLatch.await(
+                MESSAGE_TIMEOUT_MS,
+                TimeUnit.MILLISECONDS);
+            assertThat(result, is(true));
+    }
+
+    @Test
+    public void bgpOpenMessageTest7() throws InterruptedException {
+        // Open message with both LS capability and as4 capability
+        short afi = 16388;
+        byte res = 0;
+        byte safi = 71;
+        peer1.peerChannelHandler.asNumber = 200;
+        peer1.peerChannelHandler.version = 4;
+        peer1.peerChannelHandler.holdTime = 120;
+
+        peer1.peerChannelHandler.isLargeAsCapabilitySet = true;
+        bgpControllerImpl.getConfig().setLargeASCapability(true);
+        BGPValueType tempTlv = new FourOctetAsNumCapabilityTlv(200);
+        peer1.peerChannelHandler.capabilityTlv.add(tempTlv);
+
+        bgpControllerImpl.getConfig().setLsCapability(true);
+        BGPValueType tempTlv1 = new MultiProtocolExtnCapabilityTlv(afi, res, safi);
+        peer1.peerChannelHandler.capabilityTlv.add(tempTlv1);
+        peer1.connect(connectToSocket);
+
+        boolean result;
+        result = peer1.peerFrameDecoder.receivedOpenMessageLatch.await(
+            MESSAGE_TIMEOUT_MS,
+            TimeUnit.MILLISECONDS);
+        assertThat(result, is(true));
+    }
+
+    /**
+     * A class to capture the state for a BGP peer.
+     */
+    private final class BgpPeerTest {
+        private ClientBootstrap peerBootstrap;
+        private BgpPeerFrameDecoderTest peerFrameDecoder =
+                new BgpPeerFrameDecoderTest();
+        private BgpPeerChannelHandlerTest peerChannelHandler;
+
+        private BgpPeerTest(byte version, short asNumber,
+                short holdTime, int bgpId, boolean isLargeAsCapabilitySet,
+                LinkedList<BGPValueType> capabilityTlv) {
+            peerChannelHandler = new BgpPeerChannelHandlerTest(version,
+                asNumber, holdTime, bgpId, isLargeAsCapabilitySet, capabilityTlv);
+        }
+
+        /**
+         * Starts the BGP peer.
+         *
+         * @param connectToSocket the socket to connect to
+         */
+        private void connect(InetSocketAddress connectToSocket)
+            throws InterruptedException {
+
+            ChannelFactory channelFactory =
+                new NioClientSocketChannelFactory(
+                        Executors.newCachedThreadPool(),
+                        Executors.newCachedThreadPool());
+            ChannelPipelineFactory pipelineFactory = () -> {
+                ChannelPipeline pipeline = Channels.pipeline();
+                pipeline.addLast("BgpPeerFrameDecoderTest",
+                        peerFrameDecoder);
+                pipeline.addLast("BgpPeerChannelHandlerTest",
+                        peerChannelHandler);
+                return pipeline;
+            };
+
+            peerBootstrap = new ClientBootstrap(channelFactory);
+            peerBootstrap.setOption("child.keepAlive", true);
+            peerBootstrap.setOption("child.tcpNoDelay", true);
+            peerBootstrap.setPipelineFactory(pipelineFactory);
+            peerBootstrap.connect(connectToSocket);
+       }
+    }
+}
\ No newline at end of file

+/*
+ * Copyright 2014-2015 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.bgp;
+
+import java.util.LinkedList;
+import java.util.concurrent.TimeUnit;
+
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.SimpleChannelHandler;
+import org.onosproject.bgpio.protocol.ver4.BGPKeepaliveMsgVer4;
+import org.onosproject.bgpio.protocol.ver4.BGPOpenMsgVer4;
+import org.onosproject.bgpio.types.BGPHeader;
+import org.onosproject.bgpio.types.BGPValueType;
+
+public class BgpPeerChannelHandlerTest extends SimpleChannelHandler {
+    public static final int OPEN_MSG_MINIMUM_LENGTH = 29;
+    public static final byte[] MARKER = new byte[] {(byte) 0xff, (byte) 0xff,
+        (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff,
+        (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff, (byte) 0xff,
+        (byte) 0xff, (byte) 0xff};
+    public static final BGPHeader DEFAULT_OPEN_HEADER = new BGPHeader(MARKER,
+            (short) OPEN_MSG_MINIMUM_LENGTH, (byte) 0X01);
+    LinkedList<BGPValueType> capabilityTlv = new LinkedList<>();
+    public byte version;
+    public short asNumber;
+    public short holdTime;
+    public int bgpId;
+    public boolean isLargeAsCapabilitySet;
+
+    final BGPOpenMsgVer4 openMessage = new BGPOpenMsgVer4();
+    ChannelHandlerContext savedCtx;
+
+    /**
+     * Constructor to initialize all variables of BGP Open message.
+     *
+     * @param version BGP version in open message
+     * @param asNumber AS number in open message
+     * @param holdTime hold time in open message
+     * @param bgpId BGP identifier in open message
+     * @param capabilityTlv capabilities in open message
+     */
+    public BgpPeerChannelHandlerTest(byte version,
+            short asNumber,
+            short holdTime,
+            int bgpId,
+            boolean isLargeAsCapabilitySet,
+            LinkedList<BGPValueType> capabilityTlv) {
+        this.version = version;
+        this.asNumber = asNumber;
+        this.holdTime = holdTime;
+        this.bgpId = bgpId;
+        this.isLargeAsCapabilitySet = isLargeAsCapabilitySet;
+        this.capabilityTlv = capabilityTlv;
+    }
+
+    /**
+     * closes the channel.
+     */
+    void closeChannel() {
+        savedCtx.getChannel().close();
+    }
+
+    @Override
+    public void channelConnected(ChannelHandlerContext ctx,
+                                 ChannelStateEvent channelEvent) throws InterruptedException {
+        this.savedCtx = ctx;
+
+        BGPOpenMsgVer4 openMsg = new BGPOpenMsgVer4(DEFAULT_OPEN_HEADER,
+                this.version,
+                this.asNumber,
+                this.holdTime,
+                this.bgpId,
+                this.capabilityTlv);
+        ChannelBuffer buffer = ChannelBuffers.dynamicBuffer();
+        openMsg.writeTo(buffer);
+        ctx.getChannel().write(buffer);
+
+        TimeUnit.MILLISECONDS.sleep(100);
+
+        BGPKeepaliveMsgVer4 keepaliveMsg = new BGPKeepaliveMsgVer4();
+        ChannelBuffer buffer1 = ChannelBuffers.dynamicBuffer();
+        keepaliveMsg.writeTo(buffer1);
+        ctx.getChannel().write(buffer1);
+    }
+
+    @Override
+    public void channelDisconnected(ChannelHandlerContext ctx,
+                                    ChannelStateEvent channelEvent) {
+        //Do Nothing
+    }
+}

+/*
+ * Copyright 2014-2015 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.bgp;
+
+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.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.util.concurrent.CountDownLatch;
+
+/**
+ * Class to decode the message received.
+ */
+public class BgpPeerFrameDecoderTest extends FrameDecoder {
+    static final byte OPEN_MSG_TYPE = 0x1;
+    static final byte KEEPALIVE_MSG_TYPE = 0x4;
+    static final byte UPDATE_MSG_TYPE = 0x2;
+    static final byte NOTIFICATION_MSG_TYPE = 0x3;
+    static final int MINIMUM_COMMON_HEADER_LENGTH = 19;
+    static final int MINIMUM_OPEN_MSG_LENGTH = 29;
+    static final int MINIMUM_HEADER_MARKER_LENGTH = 16;
+    static final int HEADER_AND_MSG_LEN = 18;
+
+    protected static final Logger log = LoggerFactory
+            .getLogger(BgpPeerFrameDecoderTest.class);
+    final CountDownLatch receivedOpenMessageLatch = new CountDownLatch(1);
+    final CountDownLatch receivedKeepaliveMessageLatch = new CountDownLatch(1);
+    final CountDownLatch receivedNotificationMessageLatch = new CountDownLatch(1);
+
+    @Override
+    protected Object decode(ChannelHandlerContext ctx,
+                            Channel channel,
+                            ChannelBuffer cb) throws Exception {
+
+        if (cb.readableBytes() < MINIMUM_COMMON_HEADER_LENGTH) {
+            log.debug("Error: Packet length is less then minimum length");
+            return null;
+        }
+
+        byte[] marker = new byte[MINIMUM_HEADER_MARKER_LENGTH];
+        cb.readBytes(marker);
+        for (int i = 0; i < marker.length; i++) {
+            if (marker[i] != (byte) 0xff) {
+                log.debug("Error: Marker must be set all ones");
+                ctx.getChannel().close();
+                return null;
+            }
+        }
+
+        short length = cb.readShort();
+        if (length < MINIMUM_COMMON_HEADER_LENGTH) {
+            log.debug("Error: Bad message length");
+            ctx.getChannel().close();
+            return null;
+        }
+
+        if (length != (cb.readableBytes() + HEADER_AND_MSG_LEN)) {
+            log.debug("Error: Bad message length");
+            ctx.getChannel().close();
+            return null;
+        }
+
+        byte type = cb.readByte();
+        int len = length - MINIMUM_COMMON_HEADER_LENGTH;
+
+        ChannelBuffer message = cb.readBytes(len);
+
+        switch (type) {
+        case OPEN_MSG_TYPE:
+            processBgpOpen(ctx, message);
+            break;
+        case UPDATE_MSG_TYPE:
+            break;
+        case NOTIFICATION_MSG_TYPE:
+            processBgpNotification(ctx, message);
+            break;
+        case KEEPALIVE_MSG_TYPE:
+            processBgpKeepalive(ctx, message);
+            break;
+        default:
+            ctx.getChannel().close();
+            return null;
+        }
+
+        return null;
+    }
+
+    /**
+     * Processes BGP open message.
+     *
+     * @param ctx Channel handler context
+     * @param message open message
+     */
+    private void processBgpOpen(ChannelHandlerContext ctx,
+                                ChannelBuffer message) {
+        int minLength =
+            MINIMUM_OPEN_MSG_LENGTH - MINIMUM_COMMON_HEADER_LENGTH;
+        if (message.readableBytes() < minLength) {
+            log.debug("Error: Bad message length");
+            ctx.getChannel().close();
+            return;
+        }
+
+        message.readByte(); // read version
+        message.readShort(); // read AS number
+        message.readShort(); // read Hold timer
+        message.readInt(); // read BGP Identifier
+        // Optional Parameters
+        int optParamLen = message.readUnsignedByte();
+        if (message.readableBytes() < optParamLen) {
+            log.debug("Error: Bad message length");
+            ctx.getChannel().close();
+            return;
+        }
+        message.readBytes(optParamLen);
+
+        // Open message received
+        receivedOpenMessageLatch.countDown();
+    }
+
+    /**
+     * Processes BGP keepalive message.
+     *
+     * @param ctx Channel handler context
+     * @param message keepalive message
+     */
+    private void processBgpKeepalive(ChannelHandlerContext ctx,
+                                     ChannelBuffer message) {
+
+        // Keepalive message received
+        receivedKeepaliveMessageLatch.countDown();
+    }
+
+    /**
+     * Processes BGP notification message.
+     *
+     * @param ctx Channel handler context
+     * @param message notification message
+     */
+    private void processBgpNotification(ChannelHandlerContext ctx,
+                                     ChannelBuffer message) {
+        byte[] data;
+        message.readByte(); //read error code
+        message.readByte(); // read error sub code
+        if (message.readableBytes() > 0) {
+            data = new byte[message.readableBytes()];
+            message.readBytes(data, 0, message.readableBytes());
+        }
+
+        // Notification message received
+        receivedNotificationMessageLatch.countDown();
+    }
+}
\ No newline at end of file