Add CLI for viewing FPM connections.

Change-Id: I7e9e320b662a826cd2c0d49477b45110094d8e79

+/*
+ * 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 java.net.SocketAddress;
+import java.util.Map;
+
+/**
+ * Created by jono on 2/2/16.
+ */
+public interface FpmInfoService {
+
+    Map<SocketAddress, Long> peers();
+}

@@ -18,10 +18,12 @@ package org.onosproject.routing.fpm;
 
 
 import org.onosproject.routing.fpm.protocol.FpmHeader;
 import org.onosproject.routing.fpm.protocol.FpmHeader;
 
 
+import java.net.SocketAddress;
+
 /**
 /**
- * Listener for FPM messages.
+ * Listener for events from the route source.
  */
  */
-public interface FpmMessageListener {
+public interface FpmListener {
 
 
     /**
     /**
      * Handles an FPM message.
      * Handles an FPM message.
@@ -29,4 +31,19 @@ public interface FpmMessageListener {
      * @param fpmMessage FPM message
      * @param fpmMessage FPM message
      */
      */
     void fpmMessage(FpmHeader fpmMessage);
     void fpmMessage(FpmHeader fpmMessage);
+
+    /**
+     * Signifies that a new peer has attempted to initiate an FPM connection.
+     *
+     * @param address remote address of the peer
+     * @return true if the connection should be admitted, otherwise false
+     */
+    boolean peerConnected(SocketAddress address);
+
+    /**
+     * Signifies that an FPM connection has been disconnected.
+     *
+     * @param address remote address of the peer
+     */
+    void peerDisconnected(SocketAddress address);
 }
 }

@@ -15,6 +15,7 @@
  */
  */
 package org.onosproject.routing.fpm;
 package org.onosproject.routing.fpm;
 
 
+import com.google.common.collect.ImmutableMap;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -46,6 +47,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
 
 
 import java.net.InetSocketAddress;
 import java.net.InetSocketAddress;
+import java.net.SocketAddress;
 import java.util.Collections;
 import java.util.Collections;
 import java.util.Map;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentHashMap;
@@ -58,13 +60,15 @@ import static org.onlab.util.Tools.groupedThreads;
  */
  */
 @Service
 @Service
 @Component(immediate = true, enabled = false)
 @Component(immediate = true, enabled = false)
-public class FpmManager implements RouteSourceService {
+public class FpmManager implements RouteSourceService, FpmInfoService {
     private final Logger log = LoggerFactory.getLogger(getClass());
     private final Logger log = LoggerFactory.getLogger(getClass());
 
 
     private ServerBootstrap serverBootstrap;
     private ServerBootstrap serverBootstrap;
     private Channel serverChannel;
     private Channel serverChannel;
     private ChannelGroup allChannels = new DefaultChannelGroup();
     private ChannelGroup allChannels = new DefaultChannelGroup();
 
 
+    private Map<SocketAddress, Long> peers = new ConcurrentHashMap<>();
+
     private Map<IpPrefix, RouteEntry> fpmRoutes = new ConcurrentHashMap<>();
     private Map<IpPrefix, RouteEntry> fpmRoutes = new ConcurrentHashMap<>();
 
 
     private RouteListener routeListener;
     private RouteListener routeListener;
@@ -209,11 +213,31 @@ public class FpmManager implements RouteSourceService {
         routeListener.update(Collections.singletonList(routeUpdate));
         routeListener.update(Collections.singletonList(routeUpdate));
     }
     }
 
 
-    private class InternalFpmListener implements FpmMessageListener {
+    @Override
+    public Map<SocketAddress, Long> peers() {
+        return ImmutableMap.copyOf(peers);
+    }
+
+    private class InternalFpmListener implements FpmListener {
         @Override
         @Override
         public void fpmMessage(FpmHeader fpmMessage) {
         public void fpmMessage(FpmHeader fpmMessage) {
             FpmManager.this.fpmMessage(fpmMessage);
             FpmManager.this.fpmMessage(fpmMessage);
         }
         }
+
+        @Override
+        public boolean peerConnected(SocketAddress address) {
+            if (peers.keySet().contains(address)) {
+                return false;
+            }
+
+            peers.put(address, System.currentTimeMillis());
+            return true;
+        }
+
+        @Override
+        public void peerDisconnected(SocketAddress address) {
+            peers.remove(address);
+        }
     }
     }
 
 
 }
 }

@@ -35,7 +35,7 @@ public class FpmSessionHandler extends SimpleChannelHandler {
 
 
     private static Logger log = LoggerFactory.getLogger(FpmSessionHandler.class);
     private static Logger log = LoggerFactory.getLogger(FpmSessionHandler.class);
 
 
-    private final FpmMessageListener fpmListener;
+    private final FpmListener fpmListener;
 
 
     private Channel channel;
     private Channel channel;
 
 
@@ -44,7 +44,7 @@ public class FpmSessionHandler extends SimpleChannelHandler {
      *
      *
      * @param fpmListener listener for FPM messages
      * @param fpmListener listener for FPM messages
      */
      */
-    public FpmSessionHandler(FpmMessageListener fpmListener) {
+    public FpmSessionHandler(FpmListener fpmListener) {
         this.fpmListener = checkNotNull(fpmListener);
         this.fpmListener = checkNotNull(fpmListener);
     }
     }
 
 
@@ -59,26 +59,27 @@ public class FpmSessionHandler extends SimpleChannelHandler {
     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
     public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
             throws Exception {
             throws Exception {
         log.error("Exception thrown while handling FPM message", e.getCause());
         log.error("Exception thrown while handling FPM message", e.getCause());
+        if (channel != null) {
             channel.close();
             channel.close();
+        }
         handleDisconnect();
         handleDisconnect();
     }
     }
 
 
     @Override
     @Override
     public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
     public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e)
             throws Exception {
             throws Exception {
-        if (this.channel != null) {
+        if (!fpmListener.peerConnected(ctx.getChannel().getRemoteAddress())) {
             log.error("Received new FPM connection while already connected");
             log.error("Received new FPM connection while already connected");
             ctx.getChannel().close();
             ctx.getChannel().close();
             return;
             return;
         }
         }
 
 
-        this.channel = ctx.getChannel();
+        channel = ctx.getChannel();
     }
     }
 
 
     @Override
     @Override
     public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)
     public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e)
             throws Exception {
             throws Exception {
-        super.channelConnected(ctx, e);
     }
     }
 
 
     @Override
     @Override
@@ -94,6 +95,7 @@ public class FpmSessionHandler extends SimpleChannelHandler {
     }
     }
 
 
     private void handleDisconnect() {
     private void handleDisconnect() {
-        this.channel = null;
+        fpmListener.peerDisconnected(channel.getRemoteAddress());
+        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.
+ */
+
+package org.onosproject.routing.fpm.cli;
+
+import org.apache.karaf.shell.commands.Command;
+import org.onlab.util.Tools;
+import org.onosproject.cli.AbstractShellCommand;
+import org.onosproject.routing.fpm.FpmInfoService;
+
+import java.net.InetSocketAddress;
+
+/**
+ * Displays the current FPM connections.
+ */
+@Command(scope = "onos", name = "fpm-connections",
+        description = "Displays the current FPM connections")
+public class FpmConnectionsList extends AbstractShellCommand {
+
+    private static final String FORMAT = "%s:%s connected since %s";
+
+    @Override
+    protected void execute() {
+        FpmInfoService fpmInfo = get(FpmInfoService.class);
+
+        fpmInfo.peers().forEach((socketAddress, timestamp) -> {
+            if (socketAddress instanceof InetSocketAddress) {
+                InetSocketAddress inet = (InetSocketAddress) socketAddress;
+
+                print(FORMAT, inet.getHostString(), inet.getPort(), Tools.timeAgo(timestamp));
+            } else {
+                print("Unknown data format");
+            }
+        });
+    }
+}

+/*
+ * 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-related CLI commands.
+ */
+package org.onosproject.routing.fpm.cli;

@@ -49,5 +49,8 @@
     <command>
     <command>
       <action class="org.onosproject.routing.cli.RemovePeerCommand"/>
       <action class="org.onosproject.routing.cli.RemovePeerCommand"/>
     </command>
     </command>
+    <command>
+      <action class="org.onosproject.routing.fpm.cli.FpmConnectionsList"/>
+    </command>
   </command-bundle>
   </command-bundle>
 </blueprint>
 </blueprint>