mappingservice/southbound/src/main/java/org/opendaylight/lispflowmapping/southbound/LispSouthboundPlugin.java

   1 /*
   2  * Copyright (c) 2014 Contextream, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8
   9 package org.opendaylight.lispflowmapping.southbound;
  10
  11 import java.io.IOException;
  12 import java.net.DatagramPacket;
  13 import java.net.DatagramSocket;
  14 import java.net.InetAddress;
  15 import java.net.InetSocketAddress;
  16 import java.net.SocketException;
  17 import java.net.SocketTimeoutException;
  18 import java.nio.ByteBuffer;
  19
  20 import org.eclipse.osgi.framework.console.CommandProvider;
  21 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker;
  22 import org.opendaylight.controller.sal.binding.api.BindingAwareBroker.ProviderContext;
  23 import org.opendaylight.controller.sal.binding.api.BindingAwareProvider;
  24 import org.opendaylight.controller.sal.binding.api.NotificationProviderService;
  25 import org.opendaylight.controller.sal.binding.api.RpcProviderRegistry;
  26 import org.opendaylight.lispflowmapping.implementation.serializer.LispMessage;
  27 import org.opendaylight.lispflowmapping.southbound.lisp.ILispSouthboundService;
  28 import org.opendaylight.lispflowmapping.southbound.lisp.LispSouthboundService;
  29 import org.opendaylight.lispflowmapping.southbound.lisp.LispXtrSouthboundService;
  30 import org.opendaylight.lispflowmapping.type.sbplugin.IConfigLispSouthboundPlugin;
  31 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.control.plane.rev150314.LfmControlPlaneService;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.lfm.control.plane.rev150314.transportaddress.TransportAddress;
  33 import org.osgi.framework.BundleContext;
  34 import org.osgi.framework.FrameworkUtil;
  35 import org.slf4j.Logger;
  36 import org.slf4j.LoggerFactory;
  37
  38 import com.google.common.net.InetAddresses;
  39
  40 public class LispSouthboundPlugin implements IConfigLispSouthboundPlugin, CommandProvider, AutoCloseable, BindingAwareProvider {
  41     protected static final Logger LOG = LoggerFactory.getLogger(LispSouthboundPlugin.class);
  42
  43     private static Object startLock = new Object();
  44     private LispIoThread lispThread;
  45     private LispIoThread xtrThread;
  46     private LispSouthboundService lispSouthboundService;
  47     private LispXtrSouthboundService lispXtrSouthboundService;
  48     private NotificationProviderService notificationService;
  49     private RpcProviderRegistry rpcRegistry;
  50     private BindingAwareBroker broker;
  51     private volatile DatagramSocket socket = null;
  52     private volatile String bindingAddress = null;
  53     private volatile int xtrPort = LispMessage.XTR_PORT_NUM;
  54     private volatile boolean listenOnXtrPort = false;
  55     private BindingAwareBroker.RpcRegistration<LfmControlPlaneService> controlPlaneRpc;
  56     private DatagramSocket xtrSocket;
  57
  58     public void init() {
  59         LOG.info("LISP (RFC6830) Mapping Service is up!");
  60         final LfmControlPlaneRpc lfmCpRpc = new LfmControlPlaneRpc(this);
  61
  62         controlPlaneRpc = rpcRegistry.addRpcImplementation(LfmControlPlaneService.class, lfmCpRpc);
  63         broker.registerProvider(this);
  64
  65         synchronized (startLock) {
  66             lispSouthboundService = new LispSouthboundService();
  67             lispXtrSouthboundService = new LispXtrSouthboundService();
  68             registerWithOSGIConsole();
  69             lispSouthboundService.setNotificationProvider(this.notificationService);
  70             lispXtrSouthboundService.setNotificationProvider(this.notificationService);
  71             LOG.trace("Provider Session initialized");
  72             if (bindingAddress == null) {
  73                 setLispAddress("0.0.0.0");
  74             }
  75             LOG.info("LISP (RFC6830) Mapping Service is up!");
  76         }
  77     }
  78
  79     public void setNotificationProviderService(NotificationProviderService notificationService) {
  80         this.notificationService = notificationService;
  81     }
  82
  83     public void setRpcRegistryDependency(RpcProviderRegistry rpcRegistry) {
  84         this.rpcRegistry = rpcRegistry;
  85     }
  86
  87     public void setBindingAwareBroker(BindingAwareBroker broker) {
  88         this.broker = broker;
  89     }
  90
  91     private void registerWithOSGIConsole() {
  92         BundleContext bundleContext = FrameworkUtil.getBundle(this.getClass()).getBundleContext();
  93         bundleContext.registerService(CommandProvider.class.getName(), this, null);
  94         bundleContext.registerService(IConfigLispSouthboundPlugin.class.getName(), this, null);
  95     }
  96
  97     private void unloadActions() {
  98         if (lispThread != null) {
  99             lispThread.stopRunning();
 100         }
 101         lispSouthboundService = null;
 102         lispXtrSouthboundService = null;
 103         lispThread = null;
 104         xtrThread = null;
 105         bindingAddress = null;
 106         LOG.info("LISP (RFC6830) Mapping Service is down!");
 107         try {
 108             Thread.sleep(1100);
 109         } catch (InterruptedException e) {
 110         }
 111     }
 112
 113     private class LispIoThread extends Thread {
 114         private volatile boolean shouldRun;
 115         private volatile DatagramSocket threadSocket = null;
 116         private volatile ILispSouthboundService service;
 117         private volatile boolean running;
 118
 119         public LispIoThread(DatagramSocket socket, ILispSouthboundService service) {
 120             super("Lisp Thread");
 121             this.threadSocket = socket;
 122             this.service = service;
 123             shouldRun = true;
 124         }
 125
 126         @Override
 127         public void run() {
 128             running = true;
 129
 130             int lispReceiveTimeout = 1000;
 131
 132             LOG.info("LISP (RFC6830) Mapping Service is running and listening on address: " + bindingAddress
 133                     + " port: " + threadSocket.getLocalPort());
 134             try {
 135
 136                 threadSocket.setSoTimeout(lispReceiveTimeout);
 137             } catch (SocketException e) {
 138                 LOG.error("Cannot open socket on UDP port " + threadSocket.getLocalPort(), e);
 139                 return;
 140             }
 141
 142             while (shouldRun) {
 143                 byte[] buffer = new byte[4096];
 144                 DatagramPacket packet = new DatagramPacket(buffer, buffer.length);
 145                 try {
 146                     threadSocket.receive(packet);
 147                     LOG.trace("Received a packet!");
 148                 } catch (SocketTimeoutException ste) {
 149                     continue;
 150                 } catch (IOException e) {
 151                     LOG.warn("IO Exception while trying to recieve packet", e);
 152                 }
 153                 LOG.trace(String.format("Handling packet from {%s}:{%d} (len={%d})", packet.getAddress()
 154                         .getHostAddress(), packet.getPort(), packet.getLength()));
 155
 156                 try {
 157                     this.service.handlePacket(packet);
 158                 } catch (Exception e) {
 159                     LOG.warn("Error while handling packet", e);
 160                 }
 161             }
 162
 163             threadSocket.close();
 164             LOG.trace("Socket closed");
 165             running = false;
 166         }
 167
 168         public void stopRunning() {
 169             shouldRun = false;
 170         }
 171
 172         public boolean isRunning() {
 173             return running;
 174         }
 175     }
 176
 177     public static String intToIpv4(int address) {
 178         return ((address >> 24) & 0xff) + "." + //
 179                 ((address >> 16) & 0xff) + "." + //
 180                 ((address >> 8) & 0xff) + "." + //
 181                 ((address >> 0) & 0xff);
 182     }
 183
 184     public String getHelp() {
 185         StringBuffer help = new StringBuffer();
 186         help.append("---LISP Southbound Plugin---\n");
 187         return help.toString();
 188     }
 189
 190     private void startIOThread() {
 191         if (socket != null) {
 192             while (!socket.isClosed()) {
 193                 try {
 194                     Thread.sleep(500);
 195                 } catch (InterruptedException e) {
 196                 }
 197             }
 198         }
 199         try {
 200             socket = new DatagramSocket(new InetSocketAddress(bindingAddress, LispMessage.PORT_NUM));
 201             lispThread = new LispIoThread(socket, lispSouthboundService);
 202             lispThread.start();
 203             LOG.info("LISP (RFC6830) Mapping Service Southbound Plugin is up!");
 204             if (listenOnXtrPort) {
 205                 restartXtrThread();
 206             }
 207         } catch (SocketException e) {
 208             LOG.error("couldn't start socket {}", e.getMessage());
 209         }
 210     }
 211
 212     private void restartXtrThread() {
 213         try {
 214             stopXtrThread();
 215             xtrSocket = new DatagramSocket(new InetSocketAddress(bindingAddress, xtrPort));
 216             xtrThread = new LispIoThread(xtrSocket, lispXtrSouthboundService);
 217             xtrThread.start();
 218             LOG.info("xTR Southbound Plugin is up!");
 219         } catch (SocketException e) {
 220             LOG.warn("failed to start xtr thread: {}", e.getMessage());
 221         }
 222     }
 223
 224     public void handleSerializedLispBuffer(TransportAddress address, ByteBuffer outBuffer, String packetType) {
 225         DatagramPacket packet = new DatagramPacket(outBuffer.array(), outBuffer.limit());
 226         packet.setPort(address.getPort().getValue());
 227         InetAddress ip = InetAddresses.forString(address.getIpAddress().getIpv4Address().getValue());
 228         packet.setAddress(ip);
 229         try {
 230             if (LOG.isDebugEnabled()) {
 231                 LOG.trace("Sending " + packetType + " on port " + address.getPort().getValue() + " to address: " + ip);
 232             }
 233             socket.send(packet);
 234         } catch (IOException e) {
 235             LOG.warn("Failed to send " + packetType, e);
 236         }
 237     }
 238
 239     public void setLispAddress(String address) {
 240         synchronized (startLock) {
 241             if (bindingAddress != null && bindingAddress.equals(address)) {
 242                 LOG.trace("configured lisp binding address didn't change.");
 243             } else {
 244                 String action = (bindingAddress == null ? "Setting" : "Resetting");
 245                 LOG.trace(action + " lisp binding address to: " + address);
 246                 bindingAddress = address;
 247                 if (lispThread != null) {
 248                     lispThread.stopRunning();
 249                     while (lispThread.isRunning()) {
 250                         try {
 251                             Thread.sleep(500);
 252                         } catch (InterruptedException e) {
 253                         }
 254                     }
 255                 }
 256                 stopXtrThread();
 257                 startIOThread();
 258             }
 259         }
 260     }
 261
 262     private void stopXtrThread() {
 263         if (xtrThread != null) {
 264             xtrThread.stopRunning();
 265             while (xtrThread.isRunning()) {
 266                 try {
 267                     Thread.sleep(500);
 268                 } catch (InterruptedException e) {
 269                 }
 270             }
 271         }
 272     }
 273
 274     @Override
 275     public void shouldListenOnXtrPort(boolean shouldListenOnXtrPort) {
 276         listenOnXtrPort = shouldListenOnXtrPort;
 277         if (listenOnXtrPort) {
 278             LOG.debug("restarting xtr thread");
 279             restartXtrThread();
 280         } else {
 281             LOG.debug("terminating thread");
 282             stopXtrThread();
 283         }
 284     }
 285
 286     @Override
 287     public void setXtrPort(int port) {
 288         this.xtrPort = port;
 289         if (listenOnXtrPort) {
 290             restartXtrThread();
 291         }
 292     }
 293
 294     @Override
 295     public void close() throws Exception {
 296         unloadActions();
 297         controlPlaneRpc.close();
 298     }
 299
 300     @Override
 301     public void onSessionInitiated(ProviderContext session) {
 302         LOG.debug("LispSouthboundPlugin Provider Session Initiated");
 303     }
 304 }