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 LOG = 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         LOG.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             LOG.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                 LOG.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                     LOG.trace("Received a packet!");
 129                 } catch (SocketTimeoutException ste) {
 130                     continue;
 131                 } catch (IOException e) {
 132                     LOG.warn("IO Exception while trying to recieve packet", e);
 133                 }
 134                 LOG.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 (Exception e) {
 140                     LOG.warn("Error while handling packet", e);
 141                 }
 142             }
 143
 144             threadSocket.close();
 145             LOG.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             LOG.info("LISP (RFC6830) Mapping Service Southbound Plugin is up!");
 177             if (listenOnXtrPort) {
 178                 restartXtrThread();
 179             }
 180         } catch (SocketException e) {
 181             LOG.error("couldn't start socket {}", e.getMessage());
 182         }
 183     }
 184
 185     private void restartXtrThread() {
 186         try {
 187             stopXtrThread();
 188             xtrSocket = new DatagramSocket(new InetSocketAddress(bindingAddress, xtrPort));
 189             xtrThread = new LispIoThread(xtrSocket, lispXtrSouthboundService);
 190             xtrThread.start();
 191             LOG.info("xTR Southbound Plugin is up!");
 192         } catch (SocketException e) {
 193             LOG.warn("failed to start xtr thread: {}", e.getMessage());
 194         }
 195     }
 196
 197     public void onSessionInitiated(ProviderContext session) {
 198         LOG.info("LISP (RFC6830) Mapping Service is up!");
 199         synchronized (startLock) {
 200             if (!alreadyInit) {
 201                 alreadyInit = true;
 202                 lispSouthboundService = new LispSouthboundService();
 203                 lispXtrSouthboundService = new LispXtrSouthboundService();
 204                 registerWithOSGIConsole();
 205                 registerRPCs(session);
 206                 LOG.trace("Provider Session initialized");
 207                 if (bindingAddress == null) {
 208                     setLispAddress("0.0.0.0");
 209                 }
 210             }
 211
 212         }
 213     }
 214
 215     private void registerRPCs(ProviderContext session) {
 216         try {
 217             lispSouthboundService.setNotificationProvider(session.getSALService(NotificationProviderService.class));
 218             lispXtrSouthboundService.setNotificationProvider(session.getSALService(NotificationProviderService.class));
 219             session.addRpcImplementation(LispflowmappingService.class, this);
 220         } catch (Exception e) {
 221             LOG.error(e.getMessage(), e);
 222         }
 223     }
 224
 225     private void handleSerializedLispBuffer(TransportAddress address, ByteBuffer outBuffer, String packetType) {
 226         DatagramPacket packet = new DatagramPacket(outBuffer.array(), outBuffer.limit());
 227         packet.setPort(address.getPort().getValue());
 228         InetAddress ip = InetAddresses.forString(address.getIpAddress().getIpv4Address().getValue());
 229         packet.setAddress(ip);
 230         try {
 231             if (LOG.isDebugEnabled()) {
 232                 LOG.trace("Sending " + packetType + " on port " + address.getPort().getValue() + " to address: " + ip);
 233             }
 234             socket.send(packet);
 235         } catch (IOException e) {
 236             LOG.warn("Failed to send " + packetType, e);
 237         }
 238     }
 239
 240     public void setLispAddress(String address) {
 241         synchronized (startLock) {
 242             if (bindingAddress != null && bindingAddress.equals(address)) {
 243                 LOG.trace("configured lisp binding address didn't change.");
 244             } else {
 245                 String action = (bindingAddress == null ? "Setting" : "Resetting");
 246                 LOG.trace(action + " lisp binding address to: " + address);
 247                 bindingAddress = address;
 248                 if (lispThread != null) {
 249                     lispThread.stopRunning();
 250                     while (lispThread.isRunning()) {
 251                         try {
 252                             Thread.sleep(500);
 253                         } catch (InterruptedException e) {
 254                         }
 255                     }
 256                 }
 257                 stopXtrThread();
 258                 startIOThread();
 259             }
 260         }
 261     }
 262
 263     private void stopXtrThread() {
 264         if (xtrThread != null) {
 265             xtrThread.stopRunning();
 266             while (xtrThread.isRunning()) {
 267                 try {
 268                     Thread.sleep(500);
 269                 } catch (InterruptedException e) {
 270                 }
 271             }
 272         }
 273     }
 274
 275     @Override
 276     public Future<RpcResult<Void>> sendMapNotify(SendMapNotifyInput mapNotifyInput) {
 277         LOG.trace("sendMapNotify called!!");
 278         if (mapNotifyInput != null) {
 279             ByteBuffer outBuffer = MapNotifySerializer.getInstance().serialize(mapNotifyInput.getMapNotify());
 280             handleSerializedLispBuffer(mapNotifyInput.getTransportAddress(), outBuffer, MAP_NOTIFY);
 281         } else {
 282             LOG.warn("MapNotify was null");
 283         }
 284         return null;
 285     }
 286
 287     @Override
 288     public Future<RpcResult<Void>> sendMapReply(SendMapReplyInput mapReplyInput) {
 289         LOG.trace("sendMapReply called!!");
 290         if (mapReplyInput != null) {
 291             ByteBuffer outBuffer = MapReplySerializer.getInstance().serialize(mapReplyInput.getMapReply());
 292             handleSerializedLispBuffer(mapReplyInput.getTransportAddress(), outBuffer, MAP_REPlY);
 293         } else {
 294             LOG.warn("MapReply was null");
 295         }
 296         return null;
 297     }
 298
 299     @Override
 300     public Future<RpcResult<Void>> sendMapRequest(SendMapRequestInput mapRequestInput) {
 301         LOG.trace("sendMapRequest called!!");
 302         if (mapRequestInput != null) {
 303             ByteBuffer outBuffer = MapRequestSerializer.getInstance().serialize(mapRequestInput.getMapRequest());
 304             handleSerializedLispBuffer(mapRequestInput.getTransportAddress(), outBuffer, MAP_REQUEST);
 305         } else {
 306             LOG.debug("MapRequest was null");
 307         }
 308         return null;
 309     }
 310
 311     @Override
 312     public void shouldListenOnXtrPort(boolean shouldListenOnXtrPort) {
 313         listenOnXtrPort = shouldListenOnXtrPort;
 314         if (listenOnXtrPort) {
 315             LOG.debug("restarting xtr thread");
 316             restartXtrThread();
 317         } else {
 318             LOG.debug("terminating thread");
 319             stopXtrThread();
 320         }
 321     }
 322
 323     @Override
 324     public void setXtrPort(int port) {
 325         this.xtrPort = port;
 326         if (listenOnXtrPort) {
 327             restartXtrThread();
 328         }
 329     }
 330 }