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