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