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.ArrayList;
  18 import java.util.Date;
  19 import java.util.LinkedList;
  20 import java.util.List;
  21 import java.util.Queue;
  22 import java.util.concurrent.TimeUnit;
  23
  24 import org.opendaylight.protocol.framework.AbstractProtocolSession;
  25 import org.opendaylight.protocol.pcep.PCEPCloseTermination;
  26 import org.opendaylight.protocol.pcep.PCEPErrors;
  27 import org.opendaylight.protocol.pcep.PCEPMessage;
  28 import org.opendaylight.protocol.pcep.PCEPSession;
  29 import org.opendaylight.protocol.pcep.PCEPSessionListener;
  30 import org.opendaylight.protocol.pcep.PCEPTlv;
  31 import org.opendaylight.protocol.pcep.message.PCEPCloseMessage;
  32 import org.opendaylight.protocol.pcep.message.PCEPErrorMessage;
  33 import org.opendaylight.protocol.pcep.message.PCEPKeepAliveMessage;
  34 import org.opendaylight.protocol.pcep.message.PCEPOpenMessage;
  35 import org.opendaylight.protocol.pcep.object.PCEPCloseObject;
  36 import org.opendaylight.protocol.pcep.object.PCEPCloseObject.Reason;
  37 import org.opendaylight.protocol.pcep.object.PCEPErrorObject;
  38 import org.opendaylight.protocol.pcep.object.PCEPOpenObject;
  39 import org.opendaylight.protocol.pcep.tlv.NodeIdentifierTlv;
  40 import org.slf4j.Logger;
  41 import org.slf4j.LoggerFactory;
  42
  43 import com.google.common.annotations.VisibleForTesting;
  44 import com.google.common.base.Objects;
  45 import com.google.common.base.Objects.ToStringHelper;
  46 import com.google.common.base.Preconditions;
  47
  48 /**
  49  * Implementation of PCEPSession. (Not final for testing.)
  50  */
  51 class PCEPSessionImpl extends AbstractProtocolSession<PCEPMessage> implements PCEPSession, PCEPSessionRuntimeMXBean {
  52         /**
  53          * System.nanoTime value about when was sent the last message Protected to be updated also in tests.
  54          */
  55         protected volatile long lastMessageSentAt;
  56
  57         /**
  58          * System.nanoTime value about when was received the last message
  59          */
  60         private long lastMessageReceivedAt;
  61
  62         /**
  63          * Protected for testing.
  64          */
  65         protected int maxUnknownMessages;
  66
  67         protected final Queue<Long> unknownMessagesTimes = new LinkedList<Long>();
  68
  69         private final PCEPSessionListener listener;
  70
  71         /**
  72          * Open Object with session characteristics that were accepted by another PCE (sent from this session).
  73          */
  74         private final PCEPOpenObject localOpen;
  75
  76         /**
  77          * Open Object with session characteristics for this session (sent from another PCE).
  78          */
  79         private final PCEPOpenObject remoteOpen;
  80
  81         private static final Logger logger = LoggerFactory.getLogger(PCEPSessionImpl.class);
  82
  83         /**
  84          * Timer object grouping FSM Timers
  85          */
  86         private final Timer stateTimer;
  87
  88         private int sentMsgCount = 0;
  89
  90         private int receivedMsgCount = 0;
  91
  92         // True if the listener should not be notified about events
  93         private boolean closed = false;
  94
  95         private final Channel channel;
  96
  97         PCEPSessionImpl(final Timer timer, final PCEPSessionListener listener, final int maxUnknownMessages,
  98                         final Channel channel, final PCEPOpenObject localOpen, final PCEPOpenObject 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                         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                         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(Reason.EXP_DEADTIMER);
 146                         } else {
 147                                 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 (channel.isActive()) {
 169                         if (ct >= nextKeepalive) {
 170                                 this.sendMessage(new PCEPKeepAliveMessage());
 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 PCEPMessage msg) {
 190                 try {
 191                         this.channel.writeAndFlush(msg);
 192                         this.lastMessageSentAt = System.nanoTime();
 193                         if (!(msg instanceof PCEPKeepAliveMessage)) {
 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 PCEPCloseObject.Reason reason) {
 218                 logger.debug("Closing session: {}", this);
 219                 this.closed = true;
 220                 this.sendMessage(new PCEPCloseMessage(new PCEPCloseObject(reason)));
 221                 this.channel.close();
 222         }
 223
 224         private synchronized void terminate(final PCEPCloseObject.Reason reason) {
 225                 this.listener.onSessionTerminated(this, new PCEPCloseTermination(reason));
 226                 this.closed = true;
 227                 this.sendMessage(new PCEPCloseMessage(new PCEPCloseObject(reason)));
 228                 this.close();
 229         }
 230
 231         @Override
 232         public synchronized void endOfInput() {
 233                 if (!this.closed) {
 234                         this.listener.onSessionDown(this, new IOException("End of input detected. Close the session."));
 235                         this.closed = true;
 236                 }
 237         }
 238
 239         private void sendErrorMessage(final PCEPErrors value) {
 240                 this.sendErrorMessage(value, null);
 241         }
 242
 243         /**
 244          * Sends PCEP Error Message with one PCEPError and Open Object.
 245          *
 246          * @param value
 247          * @param open
 248          */
 249         private void sendErrorMessage(final PCEPErrors value, final PCEPOpenObject open) {
 250                 final PCEPErrorObject error = new PCEPErrorObject(value);
 251                 final List<PCEPErrorObject> errors = new ArrayList<PCEPErrorObject>();
 252                 errors.add(error);
 253                 this.sendMessage(new PCEPErrorMessage(open, errors, null));
 254         }
 255
 256         /**
 257          * The fact, that a message is malformed, comes from parser. In case of unrecognized message a particular error is
 258          * sent (CAPABILITY_NOT_SUPPORTED) and the method checks if the MAX_UNKNOWN_MSG per minute wasn't overstepped.
 259          * Second, any other error occurred that is specified by rfc. In this case, the an error message is generated and
 260          * sent.
 261          *
 262          * @param error documented error in RFC5440 or draft
 263          */
 264         @VisibleForTesting
 265         public void handleMalformedMessage(final PCEPErrors error) {
 266                 final long ct = System.nanoTime();
 267                 this.sendErrorMessage(error);
 268                 if (error == PCEPErrors.CAPABILITY_NOT_SUPPORTED) {
 269                         this.unknownMessagesTimes.add(ct);
 270                         while (ct - this.unknownMessagesTimes.peek() > 60 * 1E9) {
 271                                 this.unknownMessagesTimes.poll();
 272                         }
 273                         if (this.unknownMessagesTimes.size() > this.maxUnknownMessages) {
 274                                 this.terminate(Reason.TOO_MANY_UNKNOWN_MSG);
 275                         }
 276                 }
 277         }
 278
 279         /**
 280          * Handles incoming message. If the session is up, it notifies the user. The user is notified about every message
 281          * except KeepAlive.
 282          *
 283          * @param msg incoming message
 284          */
 285         @Override
 286         public void handleMessage(final PCEPMessage msg) {
 287                 // Update last reception time
 288                 this.lastMessageReceivedAt = System.nanoTime();
 289                 this.receivedMsgCount++;
 290
 291                 // Internal message handling. The user does not see these messages
 292                 if (msg instanceof PCEPKeepAliveMessage) {
 293                         // Do nothing, the timer has been already reset
 294                 } else if (msg instanceof PCEPOpenMessage) {
 295                         this.sendErrorMessage(PCEPErrors.ATTEMPT_2ND_SESSION);
 296                 } else if (msg instanceof PCEPCloseMessage) {
 297                         /*
 298                          * Session is up, we are reporting all messages to user. One notable
 299                          * exception is CLOSE message, which needs to be converted into a
 300                          * session DOWN event.
 301                          */
 302                         this.close();
 303                 } else {
 304                         // This message needs to be handled by the user
 305                         this.listener.onMessage(this, msg);
 306                 }
 307         }
 308
 309         /**
 310          * @return the sentMsgCount
 311          */
 312
 313         @Override
 314         public Integer getSentMsgCount() {
 315                 return this.sentMsgCount;
 316         }
 317
 318         /**
 319          * @return the receivedMsgCount
 320          */
 321
 322         @Override
 323         public Integer getReceivedMsgCount() {
 324                 return this.receivedMsgCount;
 325         }
 326
 327         @Override
 328         public Integer getDeadTimerValue() {
 329                 return remoteOpen.getDeadTimerValue();
 330         }
 331
 332         @Override
 333         public Integer getKeepAliveTimerValue() {
 334                 return localOpen.getKeepAliveTimerValue();
 335         }
 336
 337         @Override
 338         public String getPeerAddress() {
 339                 InetSocketAddress a = (InetSocketAddress) channel.remoteAddress();
 340                 return a.getHostName();
 341         }
 342
 343         @Override
 344         public void tearDown() {
 345                 this.close();
 346         }
 347
 348         @Override
 349         public String getNodeIdentifier() {
 350                 for (PCEPTlv tlv : this.remoteOpen.getTlvs()) {
 351                         if (tlv instanceof NodeIdentifierTlv) {
 352                                 return tlv.toString();
 353                         }
 354                 }
 355                 return "";
 356         }
 357
 358         @Override
 359         public final String toString() {
 360                 return addToStringAttributes(Objects.toStringHelper(this)).toString();
 361         }
 362
 363         protected ToStringHelper addToStringAttributes(final ToStringHelper toStringHelper) {
 364                 toStringHelper.add("localOpen", localOpen);
 365                 toStringHelper.add("remoteOpen", remoteOpen);
 366                 return toStringHelper;
 367         }
 368
 369         @Override
 370         protected void sessionUp() {
 371                 listener.onSessionUp(this);
 372         }
 373 }