netconf/netconf-client/src/main/java/org/opendaylight/netconf/client/TcpClientChannelInitializer.java

   1 /*
   2  * Copyright (c) 2014 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 package org.opendaylight.netconf.client;
   9
  10 import io.netty.channel.Channel;
  11 import io.netty.channel.ChannelHandlerContext;
  12 import io.netty.channel.ChannelOutboundHandlerAdapter;
  13 import io.netty.channel.ChannelPromise;
  14 import io.netty.channel.DefaultChannelPromise;
  15 import io.netty.util.concurrent.Future;
  16 import io.netty.util.concurrent.GenericFutureListener;
  17 import io.netty.util.concurrent.Promise;
  18 import java.net.SocketAddress;
  19
  20 final class TcpClientChannelInitializer extends AbstractClientChannelInitializer {
  21     TcpClientChannelInitializer(final NetconfClientSessionNegotiatorFactory negotiatorFactory,
  22             final NetconfClientSessionListener sessionListener) {
  23         super(negotiatorFactory, sessionListener);
  24     }
  25
  26     @Override
  27     public void initialize(final Channel ch, final Promise<NetconfClientSession> promise) {
  28         final Future<NetconfClientSession> negotiationFuture = promise;
  29
  30         //We have to add this channel outbound handler to channel pipeline, in order
  31         //to get notifications from netconf negotiatior. Set connection promise to
  32         //success only after successful negotiation.
  33         ch.pipeline().addFirst(new ChannelOutboundHandlerAdapter() {
  34             ChannelPromise connectPromise;
  35             GenericFutureListener<Future<NetconfClientSession>> negotiationFutureListener;
  36
  37             @Override
  38             public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress,
  39                                 final SocketAddress localAddress,
  40                                 final ChannelPromise channelPromise) {
  41                 connectPromise = channelPromise;
  42                 ChannelPromise tcpConnectFuture = new DefaultChannelPromise(ch);
  43
  44                 negotiationFutureListener = future -> {
  45                     if (future.isSuccess()) {
  46                         channelPromise.setSuccess();
  47                     }
  48                 };
  49
  50                 tcpConnectFuture.addListener(future -> {
  51                     if (future.isSuccess()) {
  52                         //complete connection promise with netconf negotiation future
  53                         negotiationFuture.addListener(negotiationFutureListener);
  54                     } else {
  55                         channelPromise.setFailure(future.cause());
  56                     }
  57                 });
  58                 ctx.connect(remoteAddress, localAddress, tcpConnectFuture);
  59             }
  60
  61             @Override
  62             public void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) throws Exception {
  63                 if (connectPromise == null) {
  64                     return;
  65                 }
  66
  67                 // If we have already succeeded and the session was dropped after, we need to fire inactive to notify
  68                 // reconnect logic
  69                 if (connectPromise.isSuccess()) {
  70                     ctx.fireChannelInactive();
  71                 }
  72
  73                 //If connection promise is not already set, it means negotiation failed
  74                 //we must set connection promise to failure
  75                 if (!connectPromise.isDone()) {
  76                     connectPromise.setFailure(new IllegalStateException("Negotiation failed"));
  77                 }
  78
  79                 //Remove listener from negotiation future, we don't want notifications
  80                 //from negotiation anymore
  81                 negotiationFuture.removeListener(negotiationFutureListener);
  82
  83                 super.disconnect(ctx, promise);
  84                 promise.setSuccess();
  85             }
  86         });
  87
  88         super.initialize(ch, promise);
  89     }
  90 }