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