1 files changed, 197 insertions, 0 deletions
diff --git a/src/main/java/org/onap/dmaap/dmf/mr/backends/memory/MemoryMetaBroker.java b/src/main/java/org/onap/dmaap/dmf/mr/backends/memory/MemoryMetaBroker.java
new file mode 100644
index 0000000..96d3eb5
--- /dev/null
+++ b/src/main/java/org/onap/dmaap/dmf/mr/backends/memory/MemoryMetaBroker.java
@@ -0,0 +1,197 @@
+/*******************************************************************************
+ *  ============LICENSE_START=======================================================
+ *  org.onap.dmaap
+ *  ================================================================================
+ *  Copyright © 2017 AT&T Intellectual Property. All rights reserved.
+ *  ================================================================================
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *        http://www.apache.org/licenses/LICENSE-2.0
+*  
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ *  ============LICENSE_END=========================================================
+ *  
+ *  ECOMP is a trademark and service mark of AT&T Intellectual Property.
+ *  
+ *******************************************************************************/
+package org.onap.dmaap.dmf.mr.backends.memory;
+
+import com.att.nsa.configs.ConfigDb;
+import com.att.nsa.security.NsaAcl;
+import com.att.nsa.security.NsaApiKey;
+import org.onap.dmaap.dmf.mr.metabroker.Broker;
+import org.onap.dmaap.dmf.mr.metabroker.Topic;
+
+import java.util.*;
+
+/**
+ * 
+ * @author anowarul.islam
+ *
+ */
+public class MemoryMetaBroker implements Broker {
+
+	private final MemoryQueue fQueue;
+	private final HashMap<String, MemTopic> fTopics;
+	
+	/**
+	 * 
+	 * @param mq
+	 * @param configDb
+	 * @param settings
+	 */
+	public MemoryMetaBroker(MemoryQueue mq, ConfigDb configDb) {
+	
+		fQueue = mq;
+		fTopics = new HashMap<>();
+	}
+
+	@Override
+	public List<Topic> getAllTopics() {
+		return new LinkedList<Topic>(fTopics.values());
+	}
+
+	@Override
+	public Topic getTopic(String topic) {
+		return fTopics.get(topic);
+	}
+
+	@Override
+	public Topic createTopic(String topic, String desc, String ownerApiId, int partitions, int replicas,
+			boolean transactionEnabled) throws TopicExistsException {
+		if (getTopic(topic) != null) {
+			throw new TopicExistsException(topic);
+		}
+		fQueue.createTopic(topic);
+		fTopics.put(topic, new MemTopic(topic, desc, ownerApiId, transactionEnabled));
+		return getTopic(topic);
+	}
+
+	@Override
+	public void deleteTopic(String topic) {
+		fTopics.remove(topic);
+		fQueue.removeTopic(topic);
+	}
+
+	private static class MemTopic implements Topic {
+
+		private final String fName;
+		private final String fDesc;
+		private final String fOwner;
+		private NsaAcl fReaders;
+		private NsaAcl fWriters;
+		private boolean ftransactionEnabled;
+		private String accessDenied = "User does not own this topic ";
+		
+		/**
+		 * constructor initialization
+		 * 
+		 * @param name
+		 * @param desc
+		 * @param owner
+		 * @param transactionEnabled
+		 */
+		public MemTopic(String name, String desc, String owner, boolean transactionEnabled) {
+			fName = name;
+			fDesc = desc;
+			fOwner = owner;
+			ftransactionEnabled = transactionEnabled;
+			fReaders = null;
+			fWriters = null;
+		}
+
+		@Override
+		public String getOwner() {
+			return fOwner;
+		}
+
+		@Override
+		public NsaAcl getReaderAcl() {
+			return fReaders;
+		}
+
+		@Override
+		public NsaAcl getWriterAcl() {
+			return fWriters;
+		}
+
+		@Override
+		public void checkUserRead(NsaApiKey user) throws AccessDeniedException {
+			if (fReaders != null && (user == null || !fReaders.canUser(user.getKey()))) {
+				throw new AccessDeniedException(user == null ? "" : user.getKey());
+			}
+		}
+
+		@Override
+		public void checkUserWrite(NsaApiKey user) throws AccessDeniedException {
+			if (fWriters != null && (user == null || !fWriters.canUser(user.getKey()))) {
+				throw new AccessDeniedException(user == null ? "" : user.getKey());
+			}
+		}
+
+		@Override
+		public String getName() {
+			return fName;
+		}
+
+		@Override
+		public String getDescription() {
+			return fDesc;
+		}
+
+		@Override
+		public void permitWritesFromUser(String publisherId, NsaApiKey asUser) throws AccessDeniedException {
+			if (!fOwner.equals(asUser.getKey())) {
+				throw new AccessDeniedException(accessDenied + fName);
+			}
+			if (fWriters == null) {
+				fWriters = new NsaAcl();
+			}
+			fWriters.add(publisherId);
+		}
+
+		@Override
+		public void denyWritesFromUser(String publisherId, NsaApiKey asUser) throws AccessDeniedException {
+			if (!fOwner.equals(asUser.getKey())) {
+				throw new AccessDeniedException(accessDenied + fName);
+			}
+			fWriters.remove(publisherId);
+		}
+
+		@Override
+		public void permitReadsByUser(String consumerId, NsaApiKey asUser) throws AccessDeniedException {
+			if (!fOwner.equals(asUser.getKey())) {
+				throw new AccessDeniedException(accessDenied + fName);
+			}
+			if (fReaders == null) {
+				fReaders = new NsaAcl();
+			}
+			fReaders.add(consumerId);
+		}
+
+		@Override
+		public void denyReadsByUser(String consumerId, NsaApiKey asUser) throws AccessDeniedException {
+			if (!fOwner.equals(asUser.getKey())) {
+				throw new AccessDeniedException(accessDenied + fName);
+			}
+			fReaders.remove(consumerId);
+		}
+
+		@Override
+		public boolean isTransactionEnabled() {
+			return ftransactionEnabled;
+		}
+
+		@Override
+		public Set<String> getOwners() {
+			final TreeSet<String> set = new TreeSet<> ();
+			set.add ( fOwner );
+			return set;
+		}
+	}
+}