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