Change code to use dmaap microservice

Change method calls that used to go to the dmaap adapter, to
now go to the appc service communicator, which connects to the
dmaap micro service.

Change-Id: I619fcb408e14fa0b926950192463d9c92ab7dbba
Signed-off-by: Patrick Brady <patrick.brady@att.com>
Issue-ID: APPC-1744

3 files changed, 20 insertions, 175 deletions

diff --git a/appc-oam/appc-oam-bundle/pom.xml b/appc-oam/appc-oam-bundle/pom.xml
index 5970f79e6..ef9d46d5e 100644
--- a/appc-oam/appc-oam-bundle/pom.xml
+++ b/appc-oam/appc-oam-bundle/pom.xml
@@ -175,22 +175,9 @@
             <artifactId>json</artifactId>
         </dependency>
 
-
-        <dependency>
-            <groupId>org.onap.appc</groupId>
-            <artifactId>appc-message-adapter-api</artifactId>
-            <version>${project.version}</version>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>org.onap.appc</groupId>
-            <artifactId>appc-message-adapter-factory</artifactId>
-            <version>${project.version}</version>
-            <scope>provided</scope>
-        </dependency>
         <dependency>
             <groupId>org.onap.appc</groupId>
-            <artifactId>appc-dmaap-adapter-bundle</artifactId>
+            <artifactId>appc-service-communicator-bundle</artifactId>
             <version>${project.version}</version>
             <scope>provided</scope>
         </dependency>
diff --git a/appc-oam/appc-oam-bundle/src/main/java/org/onap/appc/oam/messageadapter/MessageAdapter.java b/appc-oam/appc-oam-bundle/src/main/java/org/onap/appc/oam/messageadapter/MessageAdapter.java
index 7112b7d99..f3160511f 100644
--- a/appc-oam/appc-oam-bundle/src/main/java/org/onap/appc/oam/messageadapter/MessageAdapter.java
+++ b/appc-oam/appc-oam-bundle/src/main/java/org/onap/appc/oam/messageadapter/MessageAdapter.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * ONAP : APPC
  * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
  * ================================================================================
  * Copyright (C) 2017 Amdocs
  * =============================================================================
@@ -27,118 +27,51 @@ import com.att.eelf.configuration.EELFLogger;
 import com.att.eelf.configuration.EELFManager;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import org.apache.commons.lang.ObjectUtils;
-import org.onap.appc.adapter.message.MessageAdapterFactory;
-import org.onap.appc.adapter.message.Producer;
-import org.onap.appc.configuration.Configuration;
-import org.onap.appc.configuration.ConfigurationFactory;
-import org.osgi.framework.BundleContext;
-import org.osgi.framework.FrameworkUtil;
-import org.osgi.framework.ServiceReference;
-
-import java.util.HashSet;
-import java.util.Properties;
+import org.onap.appc.srvcomm.messaging.MessagingConnector;
 
 public class MessageAdapter {
 
     private final EELFLogger logger = EELFManager.getInstance().getLogger(MessageAdapter.class);
 
-    private final String PROP_APPC_OAM_DISABLED = "appc.OAM.disabled";
-    private final String PROP_APPC_OAM_TOPIC_WRITE = "appc.OAM.topic.write";
-    private String PROP_APPC_OAM_CLIENT_KEY = "appc.OAM.client.key";
-    private String PROP_APPC_OAM_CLIENT_SECRET = "appc.OAM.client.secret";
-    private String PROP_APPC_OAM_POOLMEMBERS = "appc.OAM.poolMembers";
+    private static final String  PROPERTIES_PREFIX = "appc.OAM";
 
-    private Producer producer;
+    private MessagingConnector messagingConnector;
     private String partition;
-    private Configuration configuration;
-    private HashSet<String> pool;
-    private String writeTopic;
-    private String apiKey;
-    private String apiSecret;
     private boolean isDisabled;
 
     /**
      * Initialize producer client to post messages using configuration properties.
      */
     public void init() {
-        configuration = ConfigurationFactory.getConfiguration();
-        Properties properties = configuration.getProperties();
-        updateProperties(properties);
 
         if (isAppcOamPropsListenerEnabled()) {
-            createProducer();
+            messagingConnector = new MessagingConnector();
         } else {
             logger.warn(String.format("The listener %s is disabled and will not be run", "appc.OAM"));
         }
     }
+    
+    public void init(MessagingConnector connector) {
 
-    /**
-     * Create producer using MessageAdapterFactory which is found through bundle context.
-     */
-    void createProducer() {
-        BundleContext ctx = FrameworkUtil.getBundle(MessageAdapter.class).getBundleContext();
-        if (ctx == null) {
-            logger.warn("MessageAdapter cannot create producer due to no bundle context.");
-            return;
-        }
-
-        ServiceReference svcRef = ctx.getServiceReference(MessageAdapterFactory.class.getName());
-        if (svcRef == null) {
-            logger.warn("MessageAdapter cannot create producer due to no MessageAdapterFactory service reference.");
-            return;
-        }
-
-        Producer localProducer = ((MessageAdapterFactory) ctx.getService(svcRef)).createProducer(pool, writeTopic,
-                apiKey, apiSecret);
-
-        for (String url : pool) {
-            if (url.contains("3905") || url.contains("https")) {
-                localProducer.useHttps(true);
-                break;
-            }
+        if (isAppcOamPropsListenerEnabled()) {
+            messagingConnector = connector;
+        } else {
+            logger.warn(String.format("The listener %s is disabled and will not be run", "appc.OAM"));
         }
-
-        producer = localProducer;
-
-        logger.debug("MessageAdapter created producer.");
     }
 
-    /**
-     * Read property value to set writeTopic, apiKey, apiSecret and pool.
-     *
-     * @param props of configuration
-     */
-    private void updateProperties(Properties props) {
-        logger.trace("Entering to updateProperties with Properties = " + ObjectUtils.toString(props));
-
-        pool = new HashSet<>();
-        if (props != null) {
-            isDisabled = Boolean.parseBoolean(props.getProperty(PROP_APPC_OAM_DISABLED));
-            writeTopic = props.getProperty(PROP_APPC_OAM_TOPIC_WRITE);
-            apiKey = props.getProperty(PROP_APPC_OAM_CLIENT_KEY);
-            apiSecret = props.getProperty(PROP_APPC_OAM_CLIENT_SECRET);
-            String hostnames = props.getProperty(PROP_APPC_OAM_POOLMEMBERS);
-            if (hostnames != null && !hostnames.isEmpty()) {
-                for (String name : hostnames.split(",")) {
-                    pool.add(name);
-                }
-            }
-        }
-    }
 
     /**
      * Get producer. If it is null, call createProducer to create it again.
      *
      * @return Producer
      */
-    Producer getProducer() {
-        if (producer == null) {
-            // In case, producer was not properly set yet, set it again.
-            logger.info("Calling createProducer as producer is null.");
-            createProducer();
+    MessagingConnector getMessagingConnector() {
+        if (messagingConnector == null) {
+            messagingConnector = new MessagingConnector();
         }
 
-        return producer;
+        return messagingConnector;
     }
 
     /**
@@ -161,8 +94,8 @@ public class MessageAdapter {
                 logger.debug("UEB Response = " + jsonMessage);
             }
 
-            Producer myProducer = getProducer();
-            success = myProducer != null && myProducer.post(this.partition, jsonMessage);
+            MessagingConnector connector = getMessagingConnector();
+            success = connector != null && connector.publishMessage(PROPERTIES_PREFIX, this.partition, jsonMessage);
         } catch (JsonProcessingException e1) {
             logger.error("Error generating Json from UEB message " + e1.getMessage());
             success = false;
diff --git a/appc-oam/appc-oam-bundle/src/test/java/org/onap/appc/oam/messageadapter/MessageAdapterTest.java b/appc-oam/appc-oam-bundle/src/test/java/org/onap/appc/oam/messageadapter/MessageAdapterTest.java
index bb73dfe8a..122ca9f96 100644
--- a/appc-oam/appc-oam-bundle/src/test/java/org/onap/appc/oam/messageadapter/MessageAdapterTest.java
+++ b/appc-oam/appc-oam-bundle/src/test/java/org/onap/appc/oam/messageadapter/MessageAdapterTest.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * ONAP : APPC
  * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2019 AT&T Intellectual Property. All rights reserved.
  * ================================================================================
  * Copyright (C) 2017 Amdocs
  * ================================================================================
@@ -29,8 +29,6 @@ import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
-import org.onap.appc.adapter.message.MessageAdapterFactory;
-import org.onap.appc.adapter.message.Producer;
 import org.onap.appc.configuration.Configuration;
 import org.mockito.Mockito;
 import org.opendaylight.yang.gen.v1.org.onap.appc.oam.rev170303.common.header.CommonHeader;
@@ -61,86 +59,16 @@ import static org.powermock.api.mockito.PowerMockito.mockStatic;
 @PrepareForTest({FrameworkUtil.class, ConfigurationFactory.class, Converter.class})
 public class MessageAdapterTest {
 
-    private Producer fakeProducer;
 
     private MessageAdapter messageAdapter;
 
 
     @Before
     public final void setup() throws Exception {
-        fakeProducer = mock(Producer.class);
         messageAdapter = new MessageAdapter();
     }
 
     @Test
-    public void testGetProducerReturnsNull() throws Exception {
-        MessageAdapter maSpy = Mockito.spy(messageAdapter);
-        Mockito.doNothing().when(maSpy).createProducer();
-
-        Producer producer = maSpy.getProducer();
-        Assert.assertTrue("getProducer() did not return null", producer == null);
-        Producer mySpyProducer = Whitebox.getInternalState(maSpy, "producer");
-        Assert.assertTrue("MessageAdapter producer is not null", mySpyProducer == null);
-        Mockito.verify(maSpy, Mockito.times(1)).createProducer();
-    }
-
-    @Test
-    public void testGetProducerWithExistingProducer() throws Exception {
-        MessageAdapter maSpy = Mockito.spy(messageAdapter);
-        Whitebox.setInternalState(maSpy, "producer", fakeProducer);
-
-        Producer producer = maSpy.getProducer();
-        Assert.assertTrue("getProducer() returned null", producer == fakeProducer);
-        Mockito.verify(maSpy, Mockito.times(0)).createProducer();
-    }
-
-    @Test
-    public void testGetProducerWithCreateProducer() throws Exception {
-        MessageAdapter maSpy = Mockito.spy(messageAdapter);
-        Whitebox.setInternalState(maSpy, "producer", (Object) null);
-        HashSet<String> pool = new HashSet<>();
-        pool.add("NOT_HTTPS");
-        pool.add("https");
-        Whitebox.setInternalState(maSpy, "pool", pool);
-
-        // Prepare all mocks
-        mockStatic(FrameworkUtil.class);
-        Bundle maBundle = mock(Bundle.class);
-        PowerMockito.when(FrameworkUtil.getBundle(MessageAdapter.class)).thenReturn(maBundle);
-
-        BundleContext maBundleContext = mock(BundleContext.class);
-        Mockito.when(maBundle.getBundleContext()).thenReturn(maBundleContext);
-
-        ServiceReference svcRef = mock(ServiceReference.class);
-        Mockito.when(maBundleContext.getServiceReference(MessageAdapterFactory.class.getName())).thenReturn(svcRef);
-
-        MessageAdapterFactory maFactory = mock(MessageAdapterFactory.class);
-        Mockito.when(maBundleContext.getService(svcRef)).thenReturn(maFactory);
-        Mockito.when(maFactory.createProducer(pool, (String) null, null, null)).thenReturn(fakeProducer);
-
-        Producer producer = maSpy.getProducer();
-        Assert.assertTrue("getProducer() result does not match", producer == fakeProducer);
-        Producer mySpyProducer = Whitebox.getInternalState(maSpy, "producer");
-        Assert.assertTrue("MessageAdapter producer does not match",mySpyProducer == fakeProducer);
-        Mockito.verify(maSpy, Mockito.times(1)).createProducer();
-    }
-
-    @Test
-    public void testUpdateProperties() {
-        MessageAdapter maSpy = Mockito.spy(messageAdapter);
-        Mockito.doNothing().when(maSpy).createProducer();
-        Whitebox.setInternalState(maSpy, "isDisabled", false);
-        PowerMockito.mockStatic(ConfigurationFactory.class);
-        Configuration mockConfig = Mockito.mock(Configuration.class);
-        Properties properties = new Properties();
-        properties.setProperty("appc.OAM.poolMembers", "hostname1,hostname2");
-        Mockito.when(ConfigurationFactory.getConfiguration()).thenReturn(mockConfig);
-        Mockito.doReturn(properties).when(mockConfig).getProperties();
-        maSpy.init();
-        assertEquals(2, ((Set<String>)Whitebox.getInternalState(maSpy, "pool")).size());
-    }
-
-    @Test
     public void testPost() throws JsonProcessingException {
         MessageAdapter maSpy = Mockito.spy(messageAdapter);
         OAMContext oamContext = new OAMContext();
@@ -158,7 +86,6 @@ public class MessageAdapterTest {
         Mockito.when(Converter.convAsyncResponseToUebOutgoingMessageJsonString(oamContext)).thenReturn("{cambriaPartition='MSO', rpcName='maintenance_mode',"
                 + " body=Body{output=MaintenanceModeOutput [_commonHeader=CommonHeader, hashCode: 14584991,"
                 + " _status=Status, hashCode: 24801521, augmentation=[]]}}");
-        Mockito.doNothing().when(maSpy).createProducer();
         maSpy.post(oamContext);
         Mockito.verify(mockLogger).trace(Mockito.contains("Entering to post"));
         Mockito.verify(mockLogger).trace("Exiting from post with (success = false)");
@@ -180,7 +107,6 @@ public class MessageAdapterTest {
         Whitebox.setInternalState(maSpy, "logger", mockLogger);
         PowerMockito.mockStatic(Converter.class);
         Mockito.when(Converter.convAsyncResponseToUebOutgoingMessageJsonString(oamContext)).thenThrow(new JsonProcessingException("ERROR") {});
-        Mockito.doNothing().when(maSpy).createProducer();
         maSpy.post(oamContext);
         Mockito.verify(mockLogger).error(Mockito.contains("Error generating Json from UEB message"));
     }
@@ -201,8 +127,7 @@ public class MessageAdapterTest {
         Whitebox.setInternalState(maSpy, "logger", mockLogger);
         PowerMockito.mockStatic(Converter.class);
         Mockito.when(Converter.convAsyncResponseToUebOutgoingMessageJsonString(oamContext)).thenThrow(new RuntimeException("ERROR"));
-        Mockito.doNothing().when(maSpy).createProducer();
         maSpy.post(oamContext);
         Mockito.verify(mockLogger).error(Mockito.contains("Error sending message to UEB ERROR"), Mockito.any(RuntimeException.class));
     }
-}
\ No newline at end of file
+}