bgpmanager/impl/src/main/java/org/opendaylight/netvirt/bgpmanager/thrift/server/BgpThriftService.java

   1 /*
   2  * Copyright (c) 2015 Ericsson India Global Services Pvt Ltd. 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.netvirt.bgpmanager.thrift.server;
  10
  11 import com.google.common.util.concurrent.ThreadFactoryBuilder;
  12 import java.net.InetSocketAddress;
  13 import java.util.concurrent.ExecutorService;
  14 import java.util.concurrent.Executors;
  15 import java.util.concurrent.Future;
  16 import java.util.concurrent.ThreadFactory;
  17 import org.apache.thrift.protocol.TBinaryProtocol;
  18 import org.apache.thrift.protocol.TProtocol;
  19 import org.apache.thrift.server.ServerContext;
  20 import org.apache.thrift.server.TServer;
  21 import org.apache.thrift.server.TServerEventHandler;
  22 import org.apache.thrift.server.TThreadedSelectorServer;
  23 import org.apache.thrift.transport.TFramedTransport;
  24 import org.apache.thrift.transport.TNonblockingServerSocket;
  25 import org.apache.thrift.transport.TNonblockingServerTransport;
  26 import org.apache.thrift.transport.TTransport;
  27 import org.apache.thrift.transport.TTransportException;
  28 import org.opendaylight.netvirt.bgpmanager.BgpConfigurationManager;
  29 import org.opendaylight.netvirt.bgpmanager.api.IBgpManager;
  30 import org.opendaylight.netvirt.bgpmanager.thrift.gen.BgpUpdater;
  31 import org.opendaylight.netvirt.bgpmanager.thrift.gen.af_afi;
  32 import org.opendaylight.netvirt.bgpmanager.thrift.gen.protocol_type;
  33 import org.opendaylight.yangtools.yang.common.Uint32;
  34 import org.slf4j.Logger;
  35 import org.slf4j.LoggerFactory;
  36
  37 public class BgpThriftService {
  38     private static final Logger LOG = LoggerFactory.getLogger(BgpThriftService.class);
  39
  40     private final InetSocketAddress bgpThriftServerSocketAddr;
  41     private final IBgpManager bgpManager;
  42     private final BgpConfigurationManager bgpConfigManager;
  43     private volatile TServer server;
  44     private static ThreadFactory thriftServerThreadFactory = new ThreadFactoryBuilder()
  45             .setNameFormat("bgp-thrift-server-%d").build();
  46     private static ExecutorService threadPool = Executors.newFixedThreadPool(1, thriftServerThreadFactory);
  47     private volatile Future ft;
  48
  49     public BgpThriftService(InetSocketAddress bgpThriftServerSocketAddr, IBgpManager bm,
  50                             BgpConfigurationManager bgpConfigManager) {
  51         this.bgpThriftServerSocketAddr = bgpThriftServerSocketAddr;
  52         bgpManager = bm;
  53         this.bgpConfigManager = bgpConfigManager;
  54     }
  55
  56     public static class ThriftClientContext implements ServerContext {
  57         TProtocol in;
  58
  59         public ThriftClientContext(TProtocol in) {
  60             this.in = in;
  61         }
  62
  63         public TProtocol getIn() {
  64             return in;
  65         }
  66     }
  67
  68     public class BgpUpdateServer implements Runnable, BgpUpdater.Iface {
  69
  70         ThriftClientContext oldThriftClientContext;
  71
  72         BgpUpdateServer() {
  73         }
  74
  75         @Override
  76         public void run() {
  77             try {
  78                 BgpUpdater.Processor processor = new BgpUpdater.Processor(this);
  79
  80                 TNonblockingServerTransport trans = new TNonblockingServerSocket(bgpThriftServerSocketAddr);
  81                 TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args(trans);
  82                 args.transportFactory(new TFramedTransport.Factory());
  83                 args.protocolFactory(new TBinaryProtocol.Factory());
  84                 args.processor(processor);
  85                 args.selectorThreads(1);
  86                 args.workerThreads(1);
  87                 ThreadFactory tf =  new ThreadFactoryBuilder().setNameFormat("bgp-receiver-%d").build();
  88                 ExecutorService service = Executors.newFixedThreadPool(1, tf);
  89                 args.executorService(service);
  90                 server = new TThreadedSelectorServer(args);
  91                 server.setServerEventHandler(new TServerEventHandler() {
  92                     @Override
  93                     public void preServe() {
  94                         LOG.info("Bgp thrift server pre serve event");
  95                     }
  96
  97                     @Override
  98                     public ServerContext createContext(TProtocol input, TProtocol output) {
  99                         LOG.info("Bgp thrift server create context event");
 100                         synchronized (this) {
 101                             if (oldThriftClientContext != null) {
 102                                 LOG.info("Bgp thrift server closing old context");
 103                                 oldThriftClientContext.getIn().getTransport().close();
 104                             } else {
 105                                 LOG.info("Bgp thrift server old context is null nothing to close");
 106                             }
 107                             oldThriftClientContext = new ThriftClientContext(input);
 108                             return oldThriftClientContext;
 109                         }
 110                     }
 111
 112                     @Override
 113                     public void deleteContext(ServerContext serverContext, TProtocol input, TProtocol output) {
 114                         synchronized (this) {
 115                             LOG.info("Bgp thrift server delete context event");
 116                             if (oldThriftClientContext == serverContext) {
 117                                 LOG.info("Bgp thrift server cleanup old context");
 118                                 oldThriftClientContext = null;
 119                             } else {
 120                                 LOG.info("Bgp thrift server cleanup context");
 121                             }
 122                         }
 123                     }
 124
 125                     @Override
 126                     public void processContext(ServerContext serverContext, TTransport inputTransport,
 127                             TTransport outputTransport) {
 128                         LOG.trace("Bgp thrift server process context event");
 129                     }
 130                 });
 131                 server.serve();
 132             } catch (TTransportException e) {
 133                 LOG.error("Exception in BGP Updater server", e);
 134             }
 135         }
 136
 137         @Override
 138         @SuppressWarnings("checkstyle:IllegalCatch")
 139         public void onUpdatePushRoute(protocol_type protocolType,
 140                                       String rd,
 141                                       String prefix,
 142                                       int plen,
 143                                       String nexthop,
 144                                       long ethtag,
 145                                       String esi,
 146                                       String macaddress,
 147                                       int l3label,
 148                                       int l2label,
 149                                       String routermac,
 150                                       af_afi afi) {
 151             LOG.debug("Update on push route : rd {} prefix {} plen {}", rd, prefix, plen);
 152
 153             // l2label is ignored even in case of RT5. only l3label considered
 154             bgpConfigManager.onUpdatePushRoute(
 155                     protocolType,
 156                     rd,
 157                     prefix,
 158                     plen,
 159                     nexthop,
 160                     macaddress,
 161                     Uint32.valueOf(l3label),
 162                     Uint32.valueOf(l2label),
 163                     routermac,
 164                     afi);
 165         }
 166
 167         @Override
 168         public void onUpdateWithdrawRoute(protocol_type protocolType,
 169                                           String rd,
 170                                           String prefix,
 171                                           int plen,
 172                                           String nexthop,
 173                                           long ethtag,
 174                                           String esi,
 175                                           String macaddress,
 176                                           int l3label,
 177                                           int l2label,
 178                                           af_afi afi) {
 179             LOG.debug("Route del ** {} ** {}/{} ", rd, prefix, plen);
 180             bgpConfigManager.onUpdateWithdrawRoute(
 181                     protocolType,
 182                     rd,
 183                     prefix,
 184                     plen,
 185                     nexthop,
 186                     macaddress);
 187         }
 188
 189         @Override
 190         public void peerDown(String ipAddress, long asNumber) {
 191             LOG.info("PeerDown {} : {} From BGP", ipAddress, asNumber);
 192             bgpConfigManager.peerDown(ipAddress, asNumber);
 193         }
 194
 195         @Override
 196         public void peerUp(String ipAddress, long asNumber) {
 197             LOG.info("PeerUp {} : {} from BGP", ipAddress, asNumber);
 198             bgpConfigManager.peerUp(ipAddress, asNumber);
 199         }
 200
 201         @Override
 202         public void onStartConfigResyncNotification() {
 203             LOG.info("BGP (re)started");
 204             bgpManager.setQbgprestartTS(System.currentTimeMillis());
 205             bgpManager.bgpRestarted();
 206         }
 207
 208         @Override
 209         public void onNotificationSendEvent(String prefix, byte errCode,
 210                 byte errSubcode) {
 211             bgpManager.sendNotificationEvent((int) errCode, (int) errSubcode);
 212         }
 213     }
 214
 215     public synchronized void start() {
 216         if (isBgpThriftServiceStarted()) {
 217             stop();
 218         }
 219         ft = threadPool.submit(new BgpUpdateServer());
 220     }
 221
 222     public boolean isBgpThriftServiceStarted() {
 223         return ((ft == null) ? false : true);
 224     }
 225
 226     public synchronized void stop() {
 227         if (server != null) {
 228             server.stop();
 229             server = null;
 230         }
 231         if (ft != null && !ft.isDone()) {
 232             ft.cancel(true);
 233         }
 234         ft = null;
 235     }
 236 }