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, final SocketAddress localAddress,
  45                                 final ChannelPromise channelPromise) throws Exception {
  46                 connectPromise = channelPromise;
  47                 ChannelPromise tcpConnectFuture = new DefaultChannelPromise(ch);
  48
  49                 negotiationFutureListener = future -> {
  50                     if (future.isSuccess()) {
  51                         connectPromise.setSuccess();
  52                     }
  53                 };
  54
  55                 tcpConnectFuture.addListener(future -> {
  56                     if(future.isSuccess()) {
  57                         //complete connection promise with netconf negotiation future
  58                         negotiationFuture.addListener(negotiationFutureListener);
  59                     } else {
  60                         connectPromise.setFailure(future.cause());
  61                     }
  62                 });
  63                 ctx.connect(remoteAddress, localAddress, tcpConnectFuture);
  64             }
  65
  66             @Override
  67             public void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) throws Exception {
  68                 // If we have already succeeded and the session was dropped after, we need to fire inactive to notify 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
  91     @Override
  92     protected void initializeSessionNegotiator(final Channel ch, final Promise<NetconfClientSession> promise) {
  93         ch.pipeline().addAfter(NETCONF_MESSAGE_DECODER, AbstractChannelInitializer.NETCONF_SESSION_NEGOTIATOR,
  94                 negotiatorFactory.getSessionNegotiator(() -> sessionListener, ch, promise));
  95     }
  96 }