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