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

   1 /*
   2  * Copyright (c) 2013 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 io.netty.channel.Channel;
  11 import io.netty.channel.ChannelFuture;
  12 import io.netty.channel.ChannelFutureListener;
  13 import io.netty.util.Timeout;
  14 import io.netty.util.Timer;
  15 import io.netty.util.TimerTask;
  16
  17 import java.io.IOException;
  18 import java.net.InetAddress;
  19 import java.net.InetSocketAddress;
  20 import java.util.Date;
  21 import java.util.LinkedList;
  22 import java.util.Queue;
  23 import java.util.concurrent.Future;
  24 import java.util.concurrent.TimeUnit;
  25
  26 import org.opendaylight.protocol.framework.AbstractProtocolSession;
  27 import org.opendaylight.protocol.pcep.PCEPCloseTermination;
  28 import org.opendaylight.protocol.pcep.PCEPErrors;
  29 import org.opendaylight.protocol.pcep.PCEPSession;
  30 import org.opendaylight.protocol.pcep.PCEPSessionListener;
  31 import org.opendaylight.protocol.pcep.TerminationReason;
  32 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.message.rev131007.CloseBuilder;
  33 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.CloseMessage;
  34 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.KeepaliveMessage;
  35 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.Message;
  36 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.OpenMessage;
  37 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.OpenObject;
  38 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.close.message.CCloseMessageBuilder;
  39 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.close.object.CCloseBuilder;
  40 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.keepalive.message.KeepaliveMessageBuilder;
  41 import org.opendaylight.yang.gen.v1.urn.opendaylight.params.xml.ns.yang.pcep.types.rev131005.open.object.Tlvs;
  42 import org.slf4j.Logger;
  43 import org.slf4j.LoggerFactory;
  44
  45 import com.google.common.annotations.VisibleForTesting;
  46 import com.google.common.base.Objects;
  47 import com.google.common.base.Objects.ToStringHelper;
  48 import com.google.common.base.Preconditions;
  49
  50 /**
  51  * Implementation of PCEPSession. (Not final for testing.)
  52  */
  53 @VisibleForTesting
  54 public class PCEPSessionImpl extends AbstractProtocolSession<Message> implements PCEPSession, PCEPSessionRuntimeMXBean {
  55         /**
  56          * System.nanoTime value about when was sent the last message Protected to be updated also in tests.
  57          */
  58         protected volatile long lastMessageSentAt;
  59
  60         /**
  61          * System.nanoTime value about when was received the last message
  62          */
  63         private long lastMessageReceivedAt;
  64
  65         /**
  66          * Protected for testing.
  67          */
  68         protected int maxUnknownMessages;
  69
  70         protected final Queue<Long> unknownMessagesTimes = new LinkedList<Long>();
  71
  72         private final PCEPSessionListener listener;
  73
  74         /**
  75          * Open Object with session characteristics that were accepted by another PCE (sent from this session).
  76          */
  77         private final OpenObject localOpen;
  78
  79         /**
  80          * Open Object with session characteristics for this session (sent from another PCE).
  81          */
  82         private final OpenObject remoteOpen;
  83
  84         private static final Logger logger = LoggerFactory.getLogger(PCEPSessionImpl.class);
  85
  86         /**
  87          * Timer object grouping FSM Timers
  88          */
  89         private final Timer stateTimer;
  90
  91         private int sentMsgCount = 0;
  92
  93         private int receivedMsgCount = 0;
  94
  95         // True if the listener should not be notified about events
  96         private boolean closed = false;
  97
  98         private final Channel channel;
  99
 100         private final KeepaliveMessage kaMessage = (KeepaliveMessage) new KeepaliveMessageBuilder().build();
 101
 102         PCEPSessionImpl(final Timer timer, final PCEPSessionListener listener, final int maxUnknownMessages, final Channel channel,
 103                         final OpenObject localOpen, final OpenObject remoteOpen) {
 104                 this.listener = Preconditions.checkNotNull(listener);
 105                 this.stateTimer = Preconditions.checkNotNull(timer);
 106                 this.channel = Preconditions.checkNotNull(channel);
 107                 this.localOpen = Preconditions.checkNotNull(localOpen);
 108                 this.remoteOpen = Preconditions.checkNotNull(remoteOpen);
 109                 this.lastMessageReceivedAt = System.nanoTime();
 110
 111                 if (this.maxUnknownMessages != 0) {
 112                         this.maxUnknownMessages = maxUnknownMessages;
 113                 }
 114
 115                 if (getDeadTimerValue() != 0) {
 116                         this.stateTimer.newTimeout(new TimerTask() {
 117                                 @Override
 118                                 public void run(final Timeout timeout) throws Exception {
 119                                         handleDeadTimer();
 120                                 }
 121                         }, getDeadTimerValue(), TimeUnit.SECONDS);
 122                 }
 123
 124                 if (getKeepAliveTimerValue() != 0) {
 125                         this.stateTimer.newTimeout(new TimerTask() {
 126                                 @Override
 127                                 public void run(final Timeout timeout) throws Exception {
 128                                         handleKeepaliveTimer();
 129                                 }
 130                         }, getKeepAliveTimerValue(), TimeUnit.SECONDS);
 131                 }
 132
 133                 logger.debug("Session started.");
 134         }
 135
 136         /**
 137          * If DeadTimer expires, the session ends. If a message (whichever) was received during this period, the DeadTimer
 138          * will be rescheduled by DEAD_TIMER_VALUE + the time that has passed from the start of the DeadTimer to the time at
 139          * which the message was received. If the session was closed by the time this method starts to execute (the session
 140          * state will become IDLE), that rescheduling won't occur.
 141          */
 142         private synchronized void handleDeadTimer() {
 143                 final long ct = System.nanoTime();
 144
 145                 final long nextDead = this.lastMessageReceivedAt + TimeUnit.SECONDS.toNanos(getDeadTimerValue());
 146
 147                 if (this.channel.isActive()) {
 148                         if (ct >= nextDead) {
 149                                 logger.debug("DeadTimer expired. " + new Date());
 150                                 this.terminate(TerminationReason.ExpDeadtimer);
 151                         } else {
 152                                 this.stateTimer.newTimeout(new TimerTask() {
 153                                         @Override
 154                                         public void run(final Timeout timeout) throws Exception {
 155                                                 handleDeadTimer();
 156                                         }
 157                                 }, nextDead - ct, TimeUnit.NANOSECONDS);
 158                         }
 159                 }
 160         }
 161
 162         /**
 163          * If KeepAlive Timer expires, sends KeepAlive message. If a message (whichever) was send during this period, the
 164          * KeepAlive Timer will be rescheduled by KEEP_ALIVE_TIMER_VALUE + the time that has passed from the start of the
 165          * KeepAlive timer to the time at which the message was sent. If the session was closed by the time this method
 166          * starts to execute (the session state will become IDLE), that rescheduling won't occur.
 167          */
 168         private synchronized void handleKeepaliveTimer() {
 169                 final long ct = System.nanoTime();
 170
 171                 long nextKeepalive = this.lastMessageSentAt + TimeUnit.SECONDS.toNanos(getKeepAliveTimerValue());
 172
 173                 if (this.channel.isActive()) {
 174                         if (ct >= nextKeepalive) {
 175                                 this.sendMessage(this.kaMessage);
 176                                 nextKeepalive = this.lastMessageSentAt + TimeUnit.SECONDS.toNanos(getKeepAliveTimerValue());
 177                         }
 178
 179                         this.stateTimer.newTimeout(new TimerTask() {
 180                                 @Override
 181                                 public void run(final Timeout timeout) throws Exception {
 182                                         handleKeepaliveTimer();
 183                                 }
 184                         }, nextKeepalive - ct, TimeUnit.NANOSECONDS);
 185                 }
 186         }
 187
 188         /**
 189          * Sends message to serialization.
 190          *
 191          * @param msg to be sent
 192          */
 193         @Override
 194         public Future<Void> sendMessage(final Message msg) {
 195                 final ChannelFuture f = this.channel.writeAndFlush(msg);
 196                 this.lastMessageSentAt = System.nanoTime();
 197                 if (!(msg instanceof KeepaliveMessage)) {
 198                         logger.debug("Message enqueued: {}", msg);
 199                 }
 200                 this.sentMsgCount++;
 201
 202                 f.addListener(new ChannelFutureListener() {
 203                         @Override
 204                         public void operationComplete(final ChannelFuture arg) {
 205                                 if (arg.isSuccess()) {
 206                                         logger.debug("Message sent to socket: {}", msg);
 207                                 } else {
 208                                         logger.debug("Message not sent: {}", msg, arg.cause());
 209                                 }
 210                         }
 211                 });
 212
 213                 return f;
 214         }
 215
 216         /**
 217          * Closes PCEP session without sending a Close message, as the channel is no longer active.
 218          */
 219         @Override
 220         public void close() {
 221                 logger.trace("Closing session: {}", this);
 222                 this.channel.close();
 223         }
 224
 225         /**
 226          * Closes PCEP session, cancels all timers, returns to state Idle, sends the Close Message. KeepAlive and DeadTimer
 227          * are cancelled if the state of the session changes to IDLE. This method is used to close the PCEP session from
 228          * inside the session or from the listener, therefore the parent of this session should be informed.
 229          */
 230         @Override
 231         public synchronized void close(final TerminationReason reason) {
 232                 logger.debug("Closing session: {}", this);
 233                 this.closed = true;
 234                 this.sendMessage(new CloseBuilder().setCCloseMessage(
 235                                 new CCloseMessageBuilder().setCClose(new CCloseBuilder().setReason(reason.getShortValue()).build()).build()).build());
 236                 this.channel.close();
 237         }
 238
 239         @Override
 240         public Tlvs getRemoteTlvs() {
 241                 return this.remoteOpen.getTlvs();
 242         }
 243
 244         @Override
 245         public InetAddress getRemoteAddress() {
 246                 return ((InetSocketAddress) this.channel.remoteAddress()).getAddress();
 247         }
 248
 249         private synchronized void terminate(final TerminationReason reason) {
 250                 this.listener.onSessionTerminated(this, new PCEPCloseTermination(reason));
 251                 this.closed = true;
 252                 this.sendMessage(new CloseBuilder().setCCloseMessage(
 253                                 new CCloseMessageBuilder().setCClose(new CCloseBuilder().setReason(reason.getShortValue()).build()).build()).build());
 254                 this.close();
 255         }
 256
 257         @Override
 258         public synchronized void endOfInput() {
 259                 if (!this.closed) {
 260                         this.listener.onSessionDown(this, new IOException("End of input detected. Close the session."));
 261                         this.closed = true;
 262                 }
 263         }
 264
 265         private void sendErrorMessage(final PCEPErrors value) {
 266                 this.sendErrorMessage(value, null);
 267         }
 268
 269         /**
 270          * Sends PCEP Error Message with one PCEPError and Open Object.
 271          *
 272          * @param value
 273          * @param open
 274          */
 275         private void sendErrorMessage(final PCEPErrors value, final OpenObject open) {
 276                 this.sendMessage(Util.createErrorMessage(value, open));
 277         }
 278
 279         /**
 280          * The fact, that a message is malformed, comes from parser. In case of unrecognized message a particular error is
 281          * sent (CAPABILITY_NOT_SUPPORTED) and the method checks if the MAX_UNKNOWN_MSG per minute wasn't overstepped.
 282          * Second, any other error occurred that is specified by rfc. In this case, the an error message is generated and
 283          * sent.
 284          *
 285          * @param error documented error in RFC5440 or draft
 286          */
 287         @VisibleForTesting
 288         public void handleMalformedMessage(final PCEPErrors error) {
 289                 final long ct = System.nanoTime();
 290                 this.sendErrorMessage(error);
 291                 if (error == PCEPErrors.CAPABILITY_NOT_SUPPORTED) {
 292                         this.unknownMessagesTimes.add(ct);
 293                         while (ct - this.unknownMessagesTimes.peek() > 60 * 1E9) {
 294                                 this.unknownMessagesTimes.poll();
 295                         }
 296                         if (this.unknownMessagesTimes.size() > this.maxUnknownMessages) {
 297                                 this.terminate(TerminationReason.TooManyUnknownMsg);
 298                         }
 299                 }
 300         }
 301
 302         /**
 303          * Handles incoming message. If the session is up, it notifies the user. The user is notified about every message
 304          * except KeepAlive.
 305          *
 306          * @param msg incoming message
 307          */
 308         @Override
 309         public void handleMessage(final Message msg) {
 310                 // Update last reception time
 311                 this.lastMessageReceivedAt = System.nanoTime();
 312                 this.receivedMsgCount++;
 313
 314                 // Internal message handling. The user does not see these messages
 315                 if (msg instanceof KeepaliveMessage) {
 316                         // Do nothing, the timer has been already reset
 317                 } else if (msg instanceof OpenMessage) {
 318                         this.sendErrorMessage(PCEPErrors.ATTEMPT_2ND_SESSION);
 319                 } else if (msg instanceof CloseMessage) {
 320                         /*
 321                          * Session is up, we are reporting all messages to user. One notable
 322                          * exception is CLOSE message, which needs to be converted into a
 323                          * session DOWN event.
 324                          */
 325                         this.close();
 326                 } else {
 327                         // This message needs to be handled by the user
 328                         this.listener.onMessage(this, msg);
 329                 }
 330         }
 331
 332         /**
 333          * @return the sentMsgCount
 334          */
 335
 336         @Override
 337         public Integer getSentMsgCount() {
 338                 return this.sentMsgCount;
 339         }
 340
 341         /**
 342          * @return the receivedMsgCount
 343          */
 344
 345         @Override
 346         public Integer getReceivedMsgCount() {
 347                 return this.receivedMsgCount;
 348         }
 349
 350         @Override
 351         public Integer getDeadTimerValue() {
 352                 return new Integer(this.remoteOpen.getDeadTimer());
 353         }
 354
 355         @Override
 356         public Integer getKeepAliveTimerValue() {
 357                 return new Integer(this.localOpen.getKeepalive());
 358         }
 359
 360         @Override
 361         public String getPeerAddress() {
 362                 final InetSocketAddress a = (InetSocketAddress) this.channel.remoteAddress();
 363                 return a.getHostName();
 364         }
 365
 366         @Override
 367         public void tearDown() {
 368                 this.close();
 369         }
 370
 371         @Override
 372         public final String toString() {
 373                 return addToStringAttributes(Objects.toStringHelper(this)).toString();
 374         }
 375
 376         protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
 377                 toStringHelper.add("localOpen", this.localOpen);
 378                 toStringHelper.add("remoteOpen", this.remoteOpen);
 379                 return toStringHelper;
 380         }
 381
 382         @Override
 383         protected void sessionUp() {
 384                 this.listener.onSessionUp(this);
 385         }
 386
 387         @Override
 388         public String getNodeIdentifier() {
 389                 if (this.remoteOpen.getTlvs() == null) {
 390                         if (this.remoteOpen.getTlvs().getPredundancyGroupId() != null) {
 391                                 return new String(this.remoteOpen.getTlvs().getPredundancyGroupId().getIdentifier());
 392                         }
 393                 }
 394                 return "";
 395         }
 396
 397 }