openflowplugin-impl/src/main/java/org/opendaylight/openflowplugin/impl/device/listener/MultiMsgCollectorImpl.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
   9 package org.opendaylight.openflowplugin.impl.device.listener;
  10
  11 import com.google.common.annotations.VisibleForTesting;
  12 import com.google.common.base.Preconditions;
  13 import com.google.common.cache.Cache;
  14 import com.google.common.cache.CacheBuilder;
  15 import com.google.common.cache.RemovalListener;
  16 import com.google.common.cache.RemovalNotification;
  17 import java.util.ArrayList;
  18 import java.util.List;
  19 import java.util.concurrent.TimeUnit;
  20 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
  21 import org.opendaylight.openflowplugin.api.openflow.device.exception.DeviceDataException;
  22 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceReplyProcessor;
  23 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
  24 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType;
  25 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
  26 import org.slf4j.Logger;
  27 import org.slf4j.LoggerFactory;
  28
  29
  30 /**
  31  * openflowplugin-api
  32  * org.opendaylight.openflowplugin.impl.openflow.device
  33  * <p>
  34  * Implementation for {@link MultiMsgCollector} interface
  35  *
  36  * @author <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
  37  * @author <a href="mailto:tkubas@cisco.com">Timotej Kubas</a>
  38  *         <p>
  39  *         Created: Mar 23, 2015
  40  */
  41 @VisibleForTesting
  42 public class MultiMsgCollectorImpl implements MultiMsgCollector {
  43
  44     private static final Logger LOG = LoggerFactory.getLogger(MultiMsgCollectorImpl.class);
  45
  46     private final Cache<Long, MultiCollectorObject> cache;
  47     private DeviceReplyProcessor deviceReplyProcessor;
  48
  49     public MultiMsgCollectorImpl() {
  50         this(DEFAULT_TIME_OUT);
  51     }
  52
  53     public MultiMsgCollectorImpl(final int timeout) {
  54         cache = initCacheBuilder(timeout).build();
  55     }
  56
  57     private static RemovalListener<Long, MultiCollectorObject> getRemovalListener() {
  58         return new RemovalListener<Long, MultiCollectorObject>() {
  59             @Override
  60             public void onRemoval(final RemovalNotification<Long, MultiCollectorObject> notification) {
  61                 LOG.trace("Removing data with XID {} from cache, cause: {}", notification.getKey(), notification.getCause());
  62                 switch (notification.getCause()) {
  63                     case EXPIRED:
  64                         notification.getValue().invalidateFutureByTimeout(notification.getKey());
  65                 }
  66             }
  67         };
  68     }
  69
  70     private static CacheBuilder<Long, MultiCollectorObject> initCacheBuilder(final int timeout) {
  71         return CacheBuilder.newBuilder()
  72                 .expireAfterAccess(timeout, TimeUnit.SECONDS)
  73                 .removalListener(getRemovalListener())
  74                 .initialCapacity(200)
  75                 .maximumSize(500)
  76                 .concurrencyLevel(1);
  77     }
  78
  79     @Override
  80     public void registerMultipartXid(final long xid) {
  81         cache.put(xid, new MultiCollectorObject());
  82     }
  83
  84     @Override
  85     public void addMultipartMsg(final MultipartReply reply) {
  86         Preconditions.checkNotNull(reply);
  87         LOG.trace("Try to add Multipart reply msg with XID {}", reply.getXid());
  88         final long xid = reply.getXid();
  89         final MultiCollectorObject cachedRef = cache.getIfPresent(xid);
  90         if (cachedRef == null) {
  91             MultipartType multipartType = reply.getType();
  92             LOG.trace("Orphaned multipart msg with XID : {} of type {}", xid, multipartType);
  93             deviceReplyProcessor.processException(new Xid(xid),
  94                     new DeviceDataException("unknown xid received for multipart of type "+multipartType));
  95             return;
  96         }
  97
  98         try {
  99             cachedRef.add(reply);
 100             LOG.trace("Multipart reply msg with XID {} added successfully.", reply.getXid());
 101             if (!reply.getFlags().isOFPMPFREQMORE()) {
 102                 // flag OFPMFFREEQMORE false says "I'm a last one'
 103                 cachedRef.publishCollection(xid); // settable future has now whole collection
 104                 cache.invalidate(xid);              // we don't need a reference anymore - remove explicitly
 105             }
 106         } catch (DeviceDataException e) {
 107             deviceReplyProcessor.processException(new Xid(xid), e);
 108         }
 109     }
 110
 111     @Override
 112     public void setDeviceReplyProcessor(final DeviceReplyProcessor deviceReplyProcessor) {
 113         this.deviceReplyProcessor = deviceReplyProcessor;
 114     }
 115
 116     private class MultiCollectorObject {
 117         private final List<MultipartReply> replyCollection;
 118         private MultipartType msgType;
 119
 120         MultiCollectorObject() {
 121             replyCollection = new ArrayList<>();
 122         }
 123
 124         void add(final MultipartReply reply) throws DeviceDataException {
 125             /* Rise possible exception if it possible */
 126             msgTypeValidation(reply.getType(), reply.getXid());
 127             replyCollection.add(reply);
 128         }
 129
 130         void publishCollection(final long xid) {
 131             deviceReplyProcessor.processReply(new Xid(xid), replyCollection);
 132         }
 133
 134         void invalidateFutureByTimeout(final long key) {
 135             final String msg = "MultiMsgCollector can not wait for last multipart any more";
 136             deviceReplyProcessor.processException(new Xid(key), new DeviceDataException(msg));
 137         }
 138
 139         private void msgTypeValidation(final MultipartType type, final long key) throws DeviceDataException {
 140             if (msgType == null) {
 141                 msgType = type;
 142                 return;
 143             }
 144             if (!msgType.equals(type)) {
 145                 final String msg = "MultiMsgCollector get incorrect multipart msg with type {}"
 146                         + " but expected type is {}";
 147                 LOG.trace(msg, type, msgType);
 148                 throw new DeviceDataException("multipart message type mismatch");
 149             }
 150         }
 151     }
 152 }