pcep/pcc-mock/src/main/java/org/opendaylight/protocol/pcep/pcc/mock/protocol/PCCDispatcherImpl.java

   1 /*
   2  * Copyright (c) 2015 Cisco Systems, 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.protocol.pcep.pcc.mock.protocol;
  10
  11 import com.google.common.base.Optional;
  12 import io.netty.bootstrap.Bootstrap;
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.ChannelInboundHandlerAdapter;
  15 import io.netty.channel.ChannelInitializer;
  16 import io.netty.channel.ChannelOption;
  17 import io.netty.channel.EventLoopGroup;
  18 import io.netty.channel.epoll.Epoll;
  19 import io.netty.channel.epoll.EpollChannelOption;
  20 import io.netty.channel.epoll.EpollEventLoopGroup;
  21 import io.netty.channel.epoll.EpollMode;
  22 import io.netty.channel.epoll.EpollSocketChannel;
  23 import io.netty.channel.nio.NioEventLoopGroup;
  24 import io.netty.channel.socket.SocketChannel;
  25 import io.netty.channel.socket.nio.NioSocketChannel;
  26 import io.netty.util.concurrent.Future;
  27 import java.math.BigInteger;
  28 import java.net.InetSocketAddress;
  29 import java.util.concurrent.ExecutionException;
  30 import javax.annotation.Nonnull;
  31 import javax.annotation.Nullable;
  32 import org.opendaylight.protocol.concepts.KeyMapping;
  33 import org.opendaylight.protocol.pcep.PCEPSession;
  34 import org.opendaylight.protocol.pcep.PCEPSessionListenerFactory;
  35 import org.opendaylight.protocol.pcep.PCEPSessionNegotiatorFactory;
  36 import org.opendaylight.protocol.pcep.impl.PCEPHandlerFactory;
  37 import org.opendaylight.protocol.pcep.pcc.mock.api.PCCDispatcher;
  38 import org.opendaylight.protocol.pcep.spi.MessageRegistry;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 public final class PCCDispatcherImpl implements PCCDispatcher, AutoCloseable {
  43
  44     private static final Logger LOG = LoggerFactory.getLogger(PCCDispatcherImpl.class);
  45
  46     private static final int CONNECT_TIMEOUT = 2000;
  47
  48     private final PCEPHandlerFactory factory;
  49     private final EventLoopGroup workerGroup;
  50
  51     public PCCDispatcherImpl(@Nonnull final MessageRegistry registry) {
  52         if (Epoll.isAvailable()) {
  53             this.workerGroup = new EpollEventLoopGroup();
  54         } else {
  55             this.workerGroup = new NioEventLoopGroup();
  56         }
  57         this.factory = new PCEPHandlerFactory(registry);
  58     }
  59
  60     @Override
  61     public Future<PCEPSession> createClient(@Nonnull final InetSocketAddress remoteAddress, final long reconnectTime,
  62         @Nonnull final PCEPSessionListenerFactory listenerFactory, @Nonnull final PCEPSessionNegotiatorFactory negotiatorFactory,
  63         @Nullable final KeyMapping keys, @Nonnull final InetSocketAddress localAddress) {
  64         return createClient(remoteAddress, reconnectTime, listenerFactory, negotiatorFactory, keys, localAddress, BigInteger.ONE);
  65     }
  66
  67     @Override
  68     public Future<PCEPSession> createClient(@Nonnull final InetSocketAddress remoteAddress, final long reconnectTime,
  69         @Nonnull final PCEPSessionListenerFactory listenerFactory, @Nonnull final PCEPSessionNegotiatorFactory negotiatorFactory,
  70         @Nullable final KeyMapping keys, @Nonnull final InetSocketAddress localAddress, @Nonnull final BigInteger dbVersion) {
  71         final Bootstrap b = new Bootstrap();
  72         b.group(this.workerGroup);
  73         b.localAddress(localAddress);
  74         final Optional<KeyMapping> optionalKey = Optional.fromNullable(keys);
  75         setChannelFactory(b, optionalKey);
  76         b.option(ChannelOption.SO_KEEPALIVE, true);
  77         b.option(ChannelOption.SO_REUSEADDR, true);
  78         b.option(ChannelOption.MAX_MESSAGES_PER_READ, 1);
  79         final long retryTimer = reconnectTime == -1 ? 0 : reconnectTime;
  80         final PCCReconnectPromise promise = new PCCReconnectPromise(remoteAddress, (int) retryTimer, CONNECT_TIMEOUT, b);
  81         final ChannelInitializer<SocketChannel> channelInitializer = new ChannelInitializer<SocketChannel>() {
  82             @Override
  83             protected void initChannel(final SocketChannel ch) throws Exception {
  84                 ch.pipeline().addLast(PCCDispatcherImpl.this.factory.getDecoders());
  85                 ch.pipeline().addLast("negotiator", negotiatorFactory.getSessionNegotiator(listenerFactory, ch, promise, new PCCPeerProposal(dbVersion)));
  86                 ch.pipeline().addLast(PCCDispatcherImpl.this.factory.getEncoders());
  87                 ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
  88                     @Override
  89                     public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
  90                         if (promise.isCancelled()) {
  91                             return;
  92                         }
  93
  94                         if (!promise.isInitialConnectFinished()) {
  95                             LOG.debug("Connection to {} was dropped during negotiation, reattempting", remoteAddress);
  96                             return;
  97                         }
  98                         LOG.debug("Reconnecting after connection to {} was dropped", remoteAddress);
  99                         PCCDispatcherImpl.this.createClient(remoteAddress, reconnectTime, listenerFactory, negotiatorFactory,
 100                             keys, localAddress, dbVersion);
 101                     }
 102                 });
 103             }
 104         };
 105         b.handler(channelInitializer);
 106         promise.connect();
 107         return promise;
 108     }
 109
 110     private static void setChannelFactory(final Bootstrap bootstrap, final Optional<KeyMapping> keys) {
 111         if (Epoll.isAvailable()) {
 112             bootstrap.channel(EpollSocketChannel.class);
 113             bootstrap.option(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
 114         } else {
 115             bootstrap.channel(NioSocketChannel.class);
 116         }
 117         if (keys.isPresent()) {
 118             if (Epoll.isAvailable()) {
 119                 bootstrap.option(EpollChannelOption.TCP_MD5SIG, keys.get());
 120             } else {
 121                 throw new UnsupportedOperationException(Epoll.unavailabilityCause().getCause());
 122             }
 123         }
 124     }
 125
 126     @Override
 127     public void close() {
 128         try {
 129             this.workerGroup.shutdownGracefully().get();
 130         } catch (final InterruptedException | ExecutionException e) {
 131             LOG.warn("Failed to properly close dispatcher.", e);
 132         }
 133     }
 134 }