1 files changed, 19 insertions, 42 deletions

diff --git a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSource.java b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSource.java
index e52204f6..f98b481f 100644
--- a/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSource.java
+++ b/policy-endpoints/src/main/java/org/onap/policy/common/endpoints/event/comm/bus/internal/SingleThreadedBusTopicSource.java
@@ -2,8 +2,9 @@
  * ============LICENSE_START=======================================================
  * policy-endpoints
  * ================================================================================
- * Copyright (C) 2017-2020 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2021 AT&T Intellectual Property. All rights reserved.
  * Modifications Copyright (C) 2018-2019 Samsung Electronics Co., Ltd.
+ * Modifications Copyright (C) 2020 Bell Canada. 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.
@@ -24,10 +25,9 @@ package org.onap.policy.common.endpoints.event.comm.bus.internal;
 import java.io.IOException;
 import java.net.MalformedURLException;
 import java.util.UUID;
-import org.onap.policy.common.endpoints.event.comm.FilterableTopicSource;
+import lombok.Getter;
 import org.onap.policy.common.endpoints.event.comm.TopicListener;
 import org.onap.policy.common.endpoints.event.comm.bus.BusTopicSource;
-import org.onap.policy.common.endpoints.event.comm.bus.internal.BusConsumer.FilterableBusConsumer;
 import org.onap.policy.common.endpoints.properties.PolicyEndPointProperties;
 import org.onap.policy.common.endpoints.utils.NetLoggerUtil;
 import org.onap.policy.common.endpoints.utils.NetLoggerUtil.EventType;
@@ -40,7 +40,7 @@ import org.slf4j.LoggerFactory;
  * notifying its listeners.
  */
 public abstract class SingleThreadedBusTopicSource extends BusTopicBase
-        implements Runnable, BusTopicSource, FilterableTopicSource {
+        implements Runnable, BusTopicSource {
 
     /**
      * Not to be converted to PolicyLogger. This will contain all instract /out traffic and only
@@ -51,21 +51,25 @@ public abstract class SingleThreadedBusTopicSource extends BusTopicBase
     /**
      * Bus consumer group.
      */
+    @Getter
     protected final String consumerGroup;
 
     /**
      * Bus consumer instance.
      */
+    @Getter
     protected final String consumerInstance;
 
     /**
      * Bus fetch timeout.
      */
+    @Getter
     protected final int fetchTimeout;
 
     /**
      * Bus fetch limit.
      */
+    @Getter
     protected final int fetchLimit;
 
     /**
@@ -86,19 +90,24 @@ public abstract class SingleThreadedBusTopicSource extends BusTopicBase
      *
      * @throws IllegalArgumentException An invalid parameter passed in
      */
-    public SingleThreadedBusTopicSource(BusTopicParams busTopicParams) {
+    protected SingleThreadedBusTopicSource(BusTopicParams busTopicParams) {
 
         super(busTopicParams);
 
-        if (busTopicParams.isConsumerGroupInvalid()) {
+        if (busTopicParams.isConsumerGroupInvalid() && busTopicParams.isConsumerInstanceInvalid()) {
             this.consumerGroup = UUID.randomUUID().toString();
-        } else {
+            this.consumerInstance = NetworkUtil.getHostname();
+
+        } else if (busTopicParams.isConsumerGroupInvalid()) {
+            this.consumerGroup = UUID.randomUUID().toString();
+            this.consumerInstance = busTopicParams.getConsumerInstance();
+
+        } else if (busTopicParams.isConsumerInstanceInvalid()) {
             this.consumerGroup = busTopicParams.getConsumerGroup();
-        }
+            this.consumerInstance = UUID.randomUUID().toString();
 
-        if (busTopicParams.isConsumerInstanceInvalid()) {
-            this.consumerInstance = NetworkUtil.getHostname();
         } else {
+            this.consumerGroup = busTopicParams.getConsumerGroup();
             this.consumerInstance = busTopicParams.getConsumerInstance();
         }
 
@@ -262,17 +271,6 @@ public abstract class SingleThreadedBusTopicSource extends BusTopicBase
         return broadcast(event);
     }
 
-
-    @Override
-    public void setFilter(String filter) {
-        if (consumer instanceof FilterableBusConsumer) {
-            ((FilterableBusConsumer) consumer).setFilter(filter);
-
-        } else {
-            throw new UnsupportedOperationException("no server-side filtering for topic " + topic);
-        }
-    }
-
     @Override
     public String toString() {
         return "SingleThreadedBusTopicSource [consumerGroup=" + consumerGroup + ", consumerInstance=" + consumerInstance
@@ -282,29 +280,8 @@ public abstract class SingleThreadedBusTopicSource extends BusTopicBase
     }
 
     @Override
-    public String getConsumerGroup() {
-        return consumerGroup;
-    }
-
-    @Override
-    public String getConsumerInstance() {
-        return consumerInstance;
-    }
-
-    @Override
     public void shutdown() {
         this.stop();
         this.topicListeners.clear();
     }
-
-    @Override
-    public int getFetchTimeout() {
-        return fetchTimeout;
-    }
-
-    @Override
-    public int getFetchLimit() {
-        return fetchLimit;
-    }
-
 }