opendaylight/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 import org.opendaylight.protocol.framework.SessionListenerFactory;
  21
  22 class TcpClientChannelInitializer extends AbstractChannelInitializer<NetconfClientSession> {
  23
  24     private final NetconfClientSessionNegotiatorFactory negotiatorFactory;
  25     private final NetconfClientSessionListener sessionListener;
  26
  27     TcpClientChannelInitializer(final NetconfClientSessionNegotiatorFactory negotiatorFactory,
  28                                 final NetconfClientSessionListener sessionListener) {
  29         this.negotiatorFactory = negotiatorFactory;
  30         this.sessionListener = sessionListener;
  31     }
  32
  33     @Override
  34     public void initialize(final Channel ch, Promise<NetconfClientSession> promise) {
  35         final Future negotiationFuture = promise;
  36
  37         //We have to add this channel outbound handler to channel pipeline, in order
  38         //to get notifications from netconf negotiatior. Set connection promise to
  39         //success only after successful negotiation.
  40         ch.pipeline().addFirst(new ChannelOutboundHandlerAdapter() {
  41             ChannelPromise connectPromise;
  42             GenericFutureListener negotiationFutureListener;
  43
  44             @Override
  45             public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress,
  46                                 final ChannelPromise channelPromise) throws Exception {
  47                 connectPromise = channelPromise;
  48                 ChannelPromise tcpConnectFuture = new DefaultChannelPromise(ch);
  49
  50                 negotiationFutureListener = new GenericFutureListener<Future<Object>>() {
  51                     @Override
  52                     public void operationComplete(Future future) throws Exception {
  53                         if (future.isSuccess())
  54                             connectPromise.setSuccess();
  55                     }
  56                 };
  57
  58                 tcpConnectFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
  59                     @Override
  60                     public void operationComplete(Future<? super Void> future) throws Exception {
  61                         if(future.isSuccess()) {
  62                             //complete connection promise with netconf negotiation future
  63                             negotiationFuture.addListener(negotiationFutureListener);
  64                         } else {
  65                             connectPromise.setFailure(future.cause());
  66                         }
  67                     }
  68                 });
  69                 ctx.connect(remoteAddress, localAddress, tcpConnectFuture);
  70             }
  71
  72             @Override
  73             public void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception {
  74                 // If we have already succeeded and the session was dropped after, we need to fire inactive to notify 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(new SessionListenerFactory<NetconfClientSessionListener>() {
 101                     @Override
 102                     public NetconfClientSessionListener getSessionListener() {
 103                         return sessionListener;
 104                     }
 105                 }, ch, promise));
 106     }
 107 }