2 * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
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
9 package org.opendaylight.openflowplugin.impl.device.listener;
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.RequestContext;
21 import org.opendaylight.openflowplugin.api.openflow.device.Xid;
22 import org.opendaylight.openflowplugin.api.openflow.device.exception.DeviceDataException;
23 import org.opendaylight.openflowplugin.api.openflow.device.handlers.DeviceReplyProcessor;
24 import org.opendaylight.openflowplugin.api.openflow.device.handlers.MultiMsgCollector;
25 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.common.types.rev130731.MultipartType;
26 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.MultipartReply;
27 import org.opendaylight.yangtools.yang.common.RpcError;
28 import org.opendaylight.yangtools.yang.common.RpcResult;
29 import org.opendaylight.yangtools.yang.common.RpcResultBuilder;
30 import org.slf4j.Logger;
31 import org.slf4j.LoggerFactory;
37 * org.opendaylight.openflowplugin.impl.openflow.device
39 * Implementation for {@link MultiMsgCollector} interface
41 * @author <a href="mailto:vdemcak@cisco.com">Vaclav Demcak</a>
42 * @author <a href="mailto:tkubas@cisco.com">Timotej Kubas</a>
44 * Created: Mar 23, 2015
47 public class MultiMsgCollectorImpl implements MultiMsgCollector {
49 private static final Logger LOG = LoggerFactory.getLogger(MultiMsgCollectorImpl.class);
51 private final Cache<Long, MultiCollectorObject> cache;
52 private DeviceReplyProcessor deviceReplyProcessor;
54 public MultiMsgCollectorImpl() {
55 this(DEFAULT_TIME_OUT);
58 public MultiMsgCollectorImpl(final int timeout) {
59 cache = initCacheBuilder(timeout).build();
62 private static RemovalListener<Long, MultiCollectorObject> getRemovalListener() {
63 return new RemovalListener<Long, MultiCollectorObject>() {
65 public void onRemoval(final RemovalNotification<Long, MultiCollectorObject> notification) {
66 LOG.trace("Removing data with XID {} from cache, cause: {}", notification.getKey(), notification.getCause());
67 switch (notification.getCause()) {
69 notification.getValue().invalidateFutureByTimeout(notification.getKey());
75 private static CacheBuilder<Long, MultiCollectorObject> initCacheBuilder(final int timeout) {
76 return CacheBuilder.newBuilder()
77 .expireAfterAccess(timeout, TimeUnit.SECONDS)
78 .removalListener(getRemovalListener())
85 public void registerMultipartRequestContext(final RequestContext<List<MultipartReply>> requestContext) {
86 cache.put(requestContext.getXid().getValue(), new MultiCollectorObject(requestContext));
90 public void addMultipartMsg(final MultipartReply reply) {
91 Preconditions.checkNotNull(reply);
92 LOG.trace("Try to add Multipart reply msg with XID {}", reply.getXid());
93 final long xid = reply.getXid();
94 final MultiCollectorObject cachedRef = cache.getIfPresent(xid);
95 if (cachedRef == null) {
96 MultipartType multipartType = reply.getType();
97 LOG.trace("Orphaned multipart msg with XID : {} of type {}", xid, multipartType);
98 deviceReplyProcessor.processException(new Xid(xid),
99 new DeviceDataException("unknown xid received for multipart of type " + multipartType));
104 cachedRef.add(reply);
105 LOG.trace("Multipart reply msg with XID {} added successfully.", reply.getXid());
106 if (!reply.getFlags().isOFPMPFREQMORE()) {
107 // flag OFPMFFREEQMORE false says "I'm a last one'
108 cachedRef.publishCollection(xid); // settable future has now whole collection
109 cache.invalidate(xid); // we don't need a reference anymore - remove explicitly
111 } catch (DeviceDataException e) {
112 deviceReplyProcessor.processException(new Xid(xid), e);
117 public void setDeviceReplyProcessor(final DeviceReplyProcessor deviceReplyProcessor) {
118 this.deviceReplyProcessor = deviceReplyProcessor;
122 public void invalidateRequestContext(final RequestContext<List<MultipartReply>> requestContext) {
123 MultiCollectorObject multiCollectorObject = cache.getIfPresent(requestContext);
124 if (null != multiCollectorObject){
125 multiCollectorObject.invalidateFutureByTimeout(requestContext.getXid().getValue());
129 private class MultiCollectorObject {
130 private final List<MultipartReply> replyCollection = new ArrayList<>();
131 private final RequestContext<List<MultipartReply>> requestContext;
132 private MultipartType msgType;
134 MultiCollectorObject(final RequestContext<List<MultipartReply>> requestContext) {
135 this.requestContext = Preconditions.checkNotNull(requestContext);
138 void add(final MultipartReply reply) throws DeviceDataException {
139 /* Rise possible exception if it possible */
140 msgTypeValidation(reply.getType(), reply.getXid());
141 replyCollection.add(reply);
144 void publishCollection(final long xid) {
145 final RpcResult<List<MultipartReply>> rpcResult = RpcResultBuilder
146 .<List<MultipartReply>>success()
147 .withResult(replyCollection)
149 requestContext.setResult(rpcResult);
151 requestContext.close();
152 } catch (final Exception e) {
153 LOG.warn("Closing RequestContext failed: {}", e.getMessage());
154 LOG.debug("Closing RequestContext failed.. ", e);
156 deviceReplyProcessor.processReply(new Xid(xid), replyCollection);
159 void invalidateFutureByTimeout(final long key) {
160 final String msg = "MultiMsgCollector can not wait for last multipart any more";
161 DeviceDataException deviceDataException = new DeviceDataException(msg);
162 final RpcResult<List<MultipartReply>> rpcResult = RpcResultBuilder
163 .<List<MultipartReply>>failed()
164 .withError(RpcError.ErrorType.APPLICATION, String.format("Message processing failed : %s", deviceDataException.getError()), deviceDataException)
166 requestContext.setResult(rpcResult);
168 requestContext.close();
169 } catch (final Exception e) {
170 LOG.warn("Closing RequestContext failed: ", e);
171 LOG.debug("Closing RequestContext failed..", e);
173 deviceReplyProcessor.processException(new Xid(key), deviceDataException);
177 private void msgTypeValidation(final MultipartType type, final long key) throws DeviceDataException {
178 if (msgType == null) {
182 if (!msgType.equals(type)) {
183 final String msg = "MultiMsgCollector get incorrect multipart msg with type {}"
184 + " but expected type is {}";
185 LOG.trace(msg, type, msgType);
186 throw new DeviceDataException("multipart message type mismatch");