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, final Promise<NetconfClientSession> promise) {
  35         final Future<NetconfClientSession> 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<Future<NetconfClientSession>> negotiationFutureListener;
  43
  44             @Override
  45             public void connect(final ChannelHandlerContext ctx, final SocketAddress remoteAddress, final SocketAddress localAddress,
  46                                 final ChannelPromise channelPromise) throws Exception {
  47                 connectPromise = channelPromise;
  48                 ChannelPromise tcpConnectFuture = new DefaultChannelPromise(ch);
  49
  50                 negotiationFutureListener = new GenericFutureListener<Future<NetconfClientSession>>() {
  51                     @Override
  52                     public void operationComplete(final Future<NetconfClientSession> future) throws Exception {
  53                         if (future.isSuccess()) {
  54                             connectPromise.setSuccess();
  55                         }
  56                     }
  57                 };
  58
  59                 tcpConnectFuture.addListener(new GenericFutureListener<Future<? super Void>>() {
  60                     @Override
  61                     public void operationComplete(final Future<? super Void> future) throws Exception {
  62                         if(future.isSuccess()) {
  63                             //complete connection promise with netconf negotiation future
  64                             negotiationFuture.addListener(negotiationFutureListener);
  65                         } else {
  66                             connectPromise.setFailure(future.cause());
  67                         }
  68                     }
  69                 });
  70                 ctx.connect(remoteAddress, localAddress, tcpConnectFuture);
  71             }
  72
  73             @Override
  74             public void disconnect(final ChannelHandlerContext ctx, final ChannelPromise promise) throws Exception {
  75                 // If we have already succeeded and the session was dropped after, we need to fire inactive to notify reconnect logic
  76                 if(connectPromise.isSuccess()) {
  77                     ctx.fireChannelInactive();
  78                 }
  79
  80                 //If connection promise is not already set, it means negotiation failed
  81                 //we must set connection promise to failure
  82                 if(!connectPromise.isDone()) {
  83                     connectPromise.setFailure(new IllegalStateException("Negotiation failed"));
  84                 }
  85
  86                 //Remove listener from negotiation future, we don't want notifications
  87                 //from negotiation anymore
  88                 negotiationFuture.removeListener(negotiationFutureListener);
  89
  90                 super.disconnect(ctx, promise);
  91                 promise.setSuccess();
  92             }
  93         });
  94
  95         super.initialize(ch, promise);
  96     }
  97
  98     @Override
  99     protected void initializeSessionNegotiator(final Channel ch, final Promise<NetconfClientSession> promise) {
 100         ch.pipeline().addAfter(NETCONF_MESSAGE_DECODER, AbstractChannelInitializer.NETCONF_SESSION_NEGOTIATOR,
 101                 negotiatorFactory.getSessionNegotiator(new SessionListenerFactory<NetconfClientSessionListener>() {
 102                     @Override
 103                     public NetconfClientSessionListener getSessionListener() {
 104                         return sessionListener;
 105                     }
 106                 }, ch, promise));
 107     }
 108 }