pcep/impl/src/main/java/org/opendaylight/protocol/pcep/impl/AbstractSessionNegotiator.java

   1 /*
   2  * Copyright (c) 2015 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.protocol.pcep.impl;
   9
  10 import com.google.common.base.Preconditions;
  11 import io.netty.channel.Channel;
  12 import io.netty.channel.ChannelFutureListener;
  13 import io.netty.channel.ChannelHandlerContext;
  14 import io.netty.channel.ChannelInboundHandlerAdapter;
  15 import io.netty.util.concurrent.Promise;
  16 import java.util.concurrent.ExecutionException;
  17 import org.opendaylight.protocol.pcep.SessionNegotiator;
  18 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.Message;
  19 import org.slf4j.Logger;
  20 import org.slf4j.LoggerFactory;
  21
  22 public abstract class AbstractSessionNegotiator extends ChannelInboundHandlerAdapter implements SessionNegotiator {
  23     private static final Logger LOG = LoggerFactory.getLogger(AbstractSessionNegotiator.class);
  24     protected final Channel channel;
  25     protected final Promise<PCEPSessionImpl> promise;
  26
  27     protected AbstractSessionNegotiator(final Promise<PCEPSessionImpl> promise, final Channel channel) {
  28         this.promise = Preconditions.checkNotNull(promise);
  29         this.channel = Preconditions.checkNotNull(channel);
  30     }
  31
  32     protected final void negotiationSuccessful(final PCEPSessionImpl session) {
  33         LOG.debug("Negotiation on channel {} successful with session {}", this.channel, session);
  34         this.channel.pipeline().replace(this, "session", session);
  35         this.promise.setSuccess(session);
  36     }
  37
  38     protected void negotiationFailed(final Throwable cause) {
  39         LOG.debug("Negotiation on channel {} failed", this.channel, cause);
  40         this.channel.close();
  41         this.promise.setFailure(cause);
  42     }
  43
  44     protected final void sendMessage(final Message msg) {
  45         this.channel.writeAndFlush(msg).addListener((ChannelFutureListener) f -> {
  46             if (!f.isSuccess()) {
  47                 LOG.info("Failed to send message {}", msg, f.cause());
  48                 negotiationFailed(f.cause());
  49             } else {
  50                 LOG.trace("Message {} sent to socket", msg);
  51             }
  52
  53         });
  54     }
  55
  56     @Override
  57     public final void channelActive(final ChannelHandlerContext ctx) {
  58         LOG.debug("Starting session negotiation on channel {}", this.channel);
  59
  60         try {
  61             this.startNegotiation();
  62         } catch (final Exception e) {
  63             LOG.warn("Unexpected negotiation failure", e);
  64             this.negotiationFailed(e);
  65         }
  66
  67     }
  68
  69     @Override
  70     public final void channelRead(final ChannelHandlerContext ctx, final Object msg) {
  71         LOG.debug("Negotiation read invoked on channel {}", this.channel);
  72
  73         try {
  74             this.handleMessage((Message) msg);
  75         } catch (Exception e) {
  76             LOG.debug("Unexpected error while handling negotiation message {}", msg, e);
  77             this.negotiationFailed(e);
  78         }
  79
  80     }
  81
  82     @Override
  83     public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cause) {
  84         LOG.info("Unexpected error during negotiation", cause);
  85         this.negotiationFailed(cause);
  86     }
  87
  88     protected abstract void startNegotiation() throws ExecutionException;
  89
  90     protected abstract void handleMessage(final Message msg);
  91 }