Add telemetry API interface for Lifecycle

Some cleanup consolidating code in cire telemetry apis.

Issue-ID: POLICY-1687
Signed-off-by: jhh <jorge.hernandez-herrero@att.com>

Change-Id: Ic5241d45eba06796192cca1e82cba5ae53dc5aaf
Signed-off-by: jhh <jorge.hernandez-herrero@att.com>

7 files changed, 247 insertions, 208 deletions

diff --git a/feature-lifecycle/src/main/feature/config/feature-lifecycle.properties b/feature-lifecycle/src/main/feature/config/feature-lifecycle.properties
index 2cdc2abf..7fd6599c 100644
--- a/feature-lifecycle/src/main/feature/config/feature-lifecycle.properties
+++ b/feature-lifecycle/src/main/feature/config/feature-lifecycle.properties
@@ -23,12 +23,10 @@ dmaap.source.topics.POLICY-PDP-PAP.servers=${env:DMAAP_SERVERS}
 dmaap.source.topics.POLICY-PDP-PAP.effectiveTopic=${env:POLICY_PDP_PAP_TOPIC}
 dmaap.source.topics.POLICY-PDP-PAP.apiKey=${env:POLICY_PDP_PAP_API_KEY}
 dmaap.source.topics.POLICY-PDP-PAP.apiSecret=${env:POLICY_PDP_PAP_API_SECRET}
-dmaap.source.topics.POLICY-PDP-PAP.managed=false
 dmaap.source.topics.POLICY-PDP-PAP.https=true
 
 dmaap.sink.topics.POLICY-PDP-PAP.servers=${env:DMAAP_SERVERS}
 dmaap.sink.topics.POLICY-PDP-PAP.effectiveTopic=${env:POLICY_PDP_PAP_TOPIC}
 dmaap.sink.topics.POLICY-PDP-PAP.apiKey=${env:POLICY_PDP_PAP_API_KEY}
 dmaap.sink.topics.POLICY-PDP-PAP.apiSecret=${env:POLICY_PDP_PAP_API_SECRET}
-dmaap.sink.topics.POLICY-PDP-PAP.managed=false
 dmaap.sink.topics.POLICY-PDP-PAP.https=true
diff --git a/feature-lifecycle/src/main/java/org/onap/policy/drools/lifecycle/LifecycleFeature.java b/feature-lifecycle/src/main/java/org/onap/policy/drools/lifecycle/LifecycleFeature.java
index 9f292946..aacd5a2e 100644
--- a/feature-lifecycle/src/main/java/org/onap/policy/drools/lifecycle/LifecycleFeature.java
+++ b/feature-lifecycle/src/main/java/org/onap/policy/drools/lifecycle/LifecycleFeature.java
@@ -34,7 +34,7 @@ public class LifecycleFeature
     /**
      * Lifecycle FSM.
      */
-    protected static final LifecycleFsm fsm = new LifecycleFsm();
+    public static final LifecycleFsm fsm = new LifecycleFsm();
 
     @Override
     public int getSequenceNumber() {
diff --git a/feature-lifecycle/src/main/java/org/onap/policy/drools/server/restful/RestLifecycleManager.java b/feature-lifecycle/src/main/java/org/onap/policy/drools/server/restful/RestLifecycleManager.java
new file mode 100644
index 00000000..e4a93310
--- /dev/null
+++ b/feature-lifecycle/src/main/java/org/onap/policy/drools/server/restful/RestLifecycleManager.java
@@ -0,0 +1,58 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 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=========================================================
+ */
+
+package org.onap.policy.drools.server.restful;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.GET;
+import javax.ws.rs.Path;
+import javax.ws.rs.Produces;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import org.onap.policy.drools.lifecycle.LifecycleFeature;
+import org.onap.policy.drools.lifecycle.LifecycleFsm;
+
+/**
+ * REST Lifecycle Manager.
+ */
+
+@Path("/policy/pdp")
+@Produces(MediaType.APPLICATION_JSON)
+@Consumes(MediaType.APPLICATION_JSON)
+@Api
+public class RestLifecycleManager {
+
+    @GET
+    @Path("engine/lifecycle/fsm")
+    @ApiOperation(value = "Retrieves the Lifecycle FSM",
+        notes = "Lifecycle FSM", response = LifecycleFsm.class)
+    public Response fsm() {
+        return Response.status(Response.Status.OK).entity(LifecycleFeature.fsm).build();
+    }
+
+    @GET
+    @Path("engine/lifecycle/fsm/state")
+    @ApiOperation(value = "Retrieves the Lifecycle FSM",
+        notes = "Lifecycle FSM", response = LifecycleFsm.class)
+    public Response state() {
+        return Response.status(Response.Status.OK).entity(LifecycleFeature.fsm.state()).build();
+    }
+
+}
diff --git a/feature-lifecycle/src/test/java/org/onap/policy/drools/lifecycle/LifecycleStatePassiveTest.java b/feature-lifecycle/src/test/java/org/onap/policy/drools/lifecycle/LifecycleStatePassiveTest.java
index b96779aa..181b11cf 100644
--- a/feature-lifecycle/src/test/java/org/onap/policy/drools/lifecycle/LifecycleStatePassiveTest.java
+++ b/feature-lifecycle/src/test/java/org/onap/policy/drools/lifecycle/LifecycleStatePassiveTest.java
@@ -66,8 +66,6 @@ public class LifecycleStatePassiveTest extends LifecycleStateRunningTest {
         fsm = new LifecycleFsm();
         fsm.setStatusTimerSeconds(15L);
         simpleStart();
-
-        assertEquals(0, fsm.client.getSink().getRecentEvents().length);
     }
 
     @Test
diff --git a/feature-lifecycle/src/test/java/org/onap/policy/drools/server/restful/RestLifecycleManagerTest.java b/feature-lifecycle/src/test/java/org/onap/policy/drools/server/restful/RestLifecycleManagerTest.java
new file mode 100644
index 00000000..a11bc020
--- /dev/null
+++ b/feature-lifecycle/src/test/java/org/onap/policy/drools/server/restful/RestLifecycleManagerTest.java
@@ -0,0 +1,90 @@
+/*-
+ * ============LICENSE_START=======================================================
+ * Copyright (C) 2019 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=========================================================
+ */
+
+package org.onap.policy.drools.server.restful;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.onap.policy.common.endpoints.event.comm.bus.internal.BusTopicParams;
+import org.onap.policy.common.endpoints.http.client.HttpClient;
+import org.onap.policy.common.endpoints.http.server.HttpServletServer;
+import org.onap.policy.common.utils.network.NetworkUtil;
+import org.onap.policy.drools.persistence.SystemPersistence;
+import org.onap.policy.models.pdp.enums.PdpState;
+
+/**
+ * REST Lifecycle Manager Test.
+ */
+public class RestLifecycleManagerTest {
+
+    /**
+     * Set up.
+     */
+    @Before
+     public void setUp() throws Exception {
+        HttpServletServer.factory.destroy();
+        HttpClient.factory.destroy();
+
+        SystemPersistence.manager.setConfigurationDir("target/test-classes");
+
+        HttpClient.factory.build(
+            BusTopicParams.builder()
+                .clientName("lifecycle")
+                .hostname("localhost")
+                .port(8765)
+                .basePath("policy/pdp/engine/lifecycle")
+                .managed(true)
+                .build());
+
+        HttpServletServer server =
+            HttpServletServer.factory.build("lifecycle", "localhost", 8765, "/", true, true);
+        server.addServletClass("/*", RestLifecycleManager.class.getName());
+        server.setSerializationProvider("org.onap.policy.common.gson.JacksonHandler");
+        server.waitedStart(5000L);
+
+        Assert.assertTrue(NetworkUtil.isTcpPortOpen("localhost", 8765, 5, 10000L));
+
+    }
+
+    /**
+     * Tear down.
+     */
+    @After
+    public void tearDown() {
+        HttpServletServer.factory.destroy();
+        HttpClient.factory.destroy();
+    }
+
+    @Test
+    public void fsm() {
+        Response response = HttpClient.factory.get("lifecycle").get("fsm");
+        assertNotNull(HttpClient.getBody(response, String.class));
+        assertEquals(Status.OK.getStatusCode(), response.getStatus());
+
+        response = HttpClient.factory.get("lifecycle").get("fsm/state");
+        assertEquals(PdpState.TERMINATED, HttpClient.getBody(response, PdpState.class));
+        assertEquals(Status.OK.getStatusCode(), response.getStatus());
+    }
+}
\ No newline at end of file
diff --git a/policy-management/src/main/java/org/onap/policy/drools/server/restful/RestManager.java b/policy-management/src/main/java/org/onap/policy/drools/server/restful/RestManager.java
index cbe2b339..30392438 100644
--- a/policy-management/src/main/java/org/onap/policy/drools/server/restful/RestManager.java
+++ b/policy-management/src/main/java/org/onap/policy/drools/server/restful/RestManager.java
@@ -49,6 +49,7 @@ import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response.Status;
+import org.onap.policy.common.endpoints.event.comm.Topic.CommInfrastructure;
 import org.onap.policy.common.endpoints.event.comm.TopicEndpoint;
 import org.onap.policy.common.endpoints.event.comm.TopicSink;
 import org.onap.policy.common.endpoints.event.comm.TopicSource;
@@ -193,7 +194,7 @@ public class RestManager {
     public Response engineUpdate(
             @ApiParam(value = "Configuration to apply", required = true) PdpdConfiguration configuration) {
         final PolicyController controller = null;
-        boolean success = true;
+        boolean success;
         try {
             success = PolicyEngine.manager.configure(configuration);
         } catch (final Exception e) {
@@ -465,7 +466,7 @@ public class RestManager {
                         .entity(new Error(controllerName + " can't be started")).build();
             }
         } catch (final IllegalStateException e) {
-            logger.info("{}: cannot start {} because of {}", this, controller, e.getMessage(), e);;
+            logger.info("{}: cannot start {} because of {}", this, controller, e.getMessage(), e);
             return Response.status(Response.Status.PARTIAL_CONTENT).entity(controller).build();
         }
 
@@ -1744,127 +1745,79 @@ public class RestManager {
     }
 
     /**
-     * GET.
-     *
-     * @return response object
+     * GET a source.
      */
     @GET
-    @Path("engine/topics/sources/ueb/{topic}")
-    @ApiOperation(value = "Retrieves an UEB managed topic source",
-            notes = "This is an UEB Network Communicaton Endpoint source of messages for the Engine",
-            response = UebTopicSource.class)
-    public Response uebSourceTopic(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getUebTopicSource(topic)).build();
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}/{topic}")
+    @ApiOperation(value = "Retrieves a managed topic source",
+            notes = "This is an Network Communication Endpoint source of messages for the Engine",
+            response = TopicSource.class)
+    public Response sourceTopic(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
+        return Response
+            .status(Response.Status.OK)
+            .entity(TopicEndpoint.manager
+                .getTopicSource(CommInfrastructure.valueOf(comm.toUpperCase()), topic))
+            .build();
     }
 
     /**
-     * GET.
-     *
-     * @return response object
+     * GET a sink.
      */
     @GET
-    @Path("engine/topics/sinks/ueb/{topic}")
-    @ApiOperation(value = "Retrieves an UEB managed topic sink",
-            notes = "This is an UEB Network Communicaton Endpoint destination of messages from the Engine",
-            response = UebTopicSink.class)
-    public Response uebSinkTopic(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getUebTopicSink(topic)).build();
+    @Path("engine/topics/sinks/{comm: ueb|dmaap|noop}/{topic}")
+    @ApiOperation(value = "Retrieves a managed topic sink",
+            notes = "This is a Network Communicaton Endpoint destination of messages from the Engine",
+            response = TopicSink.class)
+    public Response sinkTopic(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
+        return Response
+            .status(Response.Status.OK)
+            .entity(TopicEndpoint.manager
+                .getTopicSink(CommInfrastructure.valueOf(comm.toUpperCase()), topic))
+            .build();
     }
 
     /**
-     * GET.
-     *
-     * @return response object
+     * GET a source events.
      */
     @GET
-    @Path("engine/topics/sources/dmaap/{topic}")
-    @ApiOperation(value = "Retrieves a DMaaP managed topic source",
-            notes = "This is a DMaaP Network Communicaton Endpoint source of messages for the Engine",
-            response = DmaapTopicSource.class)
-    public Response dmaapSourceTopic(
-            @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getDmaapTopicSource(topic)).build();
-    }
-
-    /**
-     * GET.
-     *
-     * @return response object
-     */
-    @GET
-    @Path("engine/topics/sinks/dmaap/{topic}")
-    @ApiOperation(value = "Retrieves a DMaaP managed topic sink",
-            notes = "This is a DMaaP Network Communicaton Endpoint destination of messages from the Engine",
-            response = DmaapTopicSink.class)
-    public Response dmaapSinkTopic(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getDmaapTopicSink(topic)).build();
-    }
-
-    /**
-     * GET.
-     *
-     * @return response object
-     */
-    @GET
-    @Path("engine/topics/sources/ueb/{topic}/events")
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}/{topic}/events")
     @ApiOperation(value = "Retrieves the latest events received by an UEB topic",
-            notes = "This is a UEB Network Communicaton Endpoint source of messages for the Engine",
-            responseContainer = "List")
-    public Response uebSourceEvents(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        return Response.status(Status.OK)
-                .entity(Arrays.asList(TopicEndpoint.manager.getUebTopicSource(topic).getRecentEvents())).build();
-    }
-
-    /**
-     * GET.
-     *
-     * @return response object
-     */
-    @GET
-    @Path("engine/topics/sinks/ueb/{topic}/events")
-    @ApiOperation(value = "Retrieves the latest events sent from a topic",
-            notes = "This is a UEB Network Communicaton Endpoint sink of messages from the Engine",
+            notes = "This is a Network Communicaton Endpoint source of messages for the Engine",
             responseContainer = "List")
-    public Response uebSinkEvents(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
+    public Response sourceEvents(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
         return Response.status(Status.OK)
-                .entity(Arrays.asList(TopicEndpoint.manager.getUebTopicSink(topic).getRecentEvents())).build();
+            .entity(Arrays
+                .asList(TopicEndpoint.manager.getTopicSource(CommInfrastructure.valueOf(comm.toUpperCase()), topic)
+                        .getRecentEvents()))
+            .build();
     }
 
     /**
-     * GET.
-     *
-     * @return response object
+     * GET a sink events.
      */
     @GET
-    @Path("engine/topics/sources/dmaap/{topic}/events")
-    @ApiOperation(value = "Retrieves the latest events received by a DMaaP topic",
-            notes = "This is a DMaaP Network Communicaton Endpoint source of messages for the Engine",
-            responseContainer = "List")
-    public Response dmaapSourceEvents(
-            @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        return Response.status(Status.OK)
-                .entity(Arrays.asList(TopicEndpoint.manager.getDmaapTopicSource(topic).getRecentEvents())).build();
-    }
-
-    /**
-     * GET.
-     *
-     * @return response object
-     */
-    @GET
-    @Path("engine/topics/sinks/dmaap/{topic}/events")
-    @ApiOperation(value = "Retrieves the latest events send through a DMaaP topic",
-            notes = "This is a DMaaP Network Communicaton Endpoint destination of messages from the Engine",
-            responseContainer = "List")
-    public Response dmaapSinkEvents(@PathParam("topic") String topic) {
+    @Path("engine/topics/sinks/{comm: ueb|dmaap|noop}/{topic}/events")
+    @ApiOperation(value = "Retrieves the latest events received by an UEB topic",
+        notes = "This is a Network Communicaton Endpoint source of messages for the Engine",
+        responseContainer = "List")
+    public Response sinkEvents(
+        @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+        @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
         return Response.status(Status.OK)
-                .entity(Arrays.asList(TopicEndpoint.manager.getDmaapTopicSink(topic).getRecentEvents())).build();
+            .entity(Arrays
+                .asList(TopicEndpoint.manager.getTopicSink(CommInfrastructure.valueOf(comm.toUpperCase()), topic)
+                    .getRecentEvents()))
+            .build();
     }
 
     /**
-     * GET.
-     *
-     * @return response object
+     * GET noop sinks.
      */
     @GET
     @Path("engine/topics/sinks/noop")
@@ -1880,33 +1833,6 @@ public class RestManager {
      * @return response object
      */
     @GET
-    @Path("engine/topics/sinks/noop/{topic}")
-    @ApiOperation(value = "Retrieves a NOOP managed topic sink",
-            notes = "NOOP is an dev/null Network Communicaton Sink", response = NoopTopicSink.class)
-    public Response noopSinkTopic(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic) {
-        return Response.status(Response.Status.OK).entity(TopicEndpoint.manager.getNoopTopicSink(topic)).build();
-    }
-
-    /**
-     * GET.
-     *
-     * @return response object
-     */
-    @GET
-    @Path("engine/topics/sinks/noop/{topic}/events")
-    @ApiOperation(value = "Retrieves the latest events send through a NOOP topic",
-            notes = "NOOP is an dev/null Network Communicaton Sink", responseContainer = "List")
-    public Response noopSinkEvents(@PathParam("topic") String topic) {
-        return Response.status(Status.OK)
-                .entity(Arrays.asList(TopicEndpoint.manager.getNoopTopicSink(topic).getRecentEvents())).build();
-    }
-
-    /**
-     * GET.
-     *
-     * @return response object
-     */
-    @GET
     @Path("engine/topics/sources/ueb/{topic}/switches")
     @ApiOperation(value = "UEB Topic Control Switches", notes = "List of the UEB Topic Control Switches",
             responseContainer = "List")
@@ -2013,93 +1939,49 @@ public class RestManager {
     }
 
     /**
-     * PUT.
+     * Offers and event to a topic in a communication infrastructure.
      *
      * @return response object
      */
     @PUT
-    @Path("engine/topics/sources/ueb/{topic}/events")
+    @Path("engine/topics/sources/{comm: ueb|dmaap|noop}/{topic}/events")
     @Consumes(MediaType.TEXT_PLAIN)
-    @ApiOperation(value = "Offers an event to an UEB topic for internal processing by the engine",
+    @ApiOperation(value = "Offers an event to a topic for internal processing by the engine",
             notes = "The offered event is treated as it was incoming from the network", responseContainer = "List")
     @ApiResponses(value = {@ApiResponse(code = 404, message = "The topic information cannot be found"),
             @ApiResponse(code = 406,
                     message = "The system is an administrative state that prevents " + "this request to be fulfilled"),
             @ApiResponse(code = 500, message = "A server error has occurred processing this request")})
-    public Response uebOffer(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic,
+    public Response commEventOffer(
+            @ApiParam(value = "Communication Mechanism", required = true) @PathParam("comm") String comm,
+            @ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic,
             @ApiParam(value = "Network Message", required = true) String json) {
-        try {
-            final UebTopicSource uebReader = TopicEndpoint.manager.getUebTopicSource(topic);
-            final boolean success = uebReader.offer(json);
-            if (success) {
-                return Response.status(Status.OK)
-                        .entity(Arrays.asList(TopicEndpoint.manager.getUebTopicSource(topic).getRecentEvents()))
-                        .build();
-            } else {
-                return Response.status(Status.NOT_ACCEPTABLE).entity(new Error("Failure to inject event over " + topic))
-                        .build();
-            }
-        } catch (final IllegalArgumentException e) {
-            logNoUebEncoder(topic, e);
-            return Response.status(Response.Status.NOT_FOUND).entity(new Error(topic + " not found")).build();
-        } catch (final IllegalStateException e) {
-            logNoUebEncoder(topic, e);
-            return Response.status(Response.Status.NOT_ACCEPTABLE)
-                    .entity(new Error(topic + " not acceptable due to current state")).build();
-        } catch (final Exception e) {
-            logNoUebEncoder(topic, e);
-            return Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(new Error(e.getMessage())).build();
-        }
-    }
-
-    private void logNoUebEncoder(String topic, Exception ex) {
-        logger.debug("{}: cannot offer for encoder ueb topic for {} because of {}", this, topic, ex.getMessage(), ex);
-    }
 
-    /**
-     * PUT.
-     *
-     * @return response object
-     */
-    @PUT
-    @Path("engine/topics/sources/dmaap/{topic}/events")
-    @Consumes(MediaType.TEXT_PLAIN)
-    @ApiOperation(value = "Offers an event to a DMaaP topic for internal processing by the engine",
-            notes = "The offered event is treated as it was incoming from the network", responseContainer = "List")
-    @ApiResponses(value = {@ApiResponse(code = 404, message = "The topic information cannot be found"),
-            @ApiResponse(code = 406,
-                    message = "The system is an administrative state that prevents " + "this request to be fulfilled"),
-            @ApiResponse(code = 500, message = "A server error has occurred processing this request")})
-    public Response dmaapOffer(@ApiParam(value = "Topic Name", required = true) @PathParam("topic") String topic,
-            @ApiParam(value = "Network Message", required = true) String json) {
         try {
-            final DmaapTopicSource dmaapReader = TopicEndpoint.manager.getDmaapTopicSource(topic);
-            final boolean success = dmaapReader.offer(json);
-            if (success) {
+            TopicSource source =
+                TopicEndpoint.manager.getTopicSource(CommInfrastructure.valueOf(comm.toUpperCase()), topic);
+            if (source.offer(json)) {
                 return Response.status(Status.OK)
-                        .entity(Arrays.asList(TopicEndpoint.manager.getDmaapTopicSource(topic).getRecentEvents()))
-                        .build();
+                    .entity(Arrays.asList(source.getRecentEvents()))
+                    .build();
             } else {
-                return Response.status(Status.NOT_ACCEPTABLE).entity(new Error("Failure to inject event over " + topic))
-                        .build();
+                return Response.status(Status.NOT_ACCEPTABLE)
+                    .entity(new Error("Failure to inject event over " + topic))
+                    .build();
             }
-        } catch (final IllegalArgumentException e) {
-            logNoDmaapEncoder(topic, e);
-            return Response.status(Response.Status.NOT_FOUND).entity(new Error(topic + " not found")).build();
-        } catch (final IllegalStateException e) {
-            logNoDmaapEncoder(topic, e);
+        } catch (IllegalArgumentException e) {
+            return Response.status(Response.Status.NOT_FOUND)
+                    .entity(new Error(topic + " not found")).build();
+        } catch (IllegalStateException e) {
             return Response.status(Response.Status.NOT_ACCEPTABLE)
-                    .entity(new Error(topic + " not acceptable due to current state")).build();
-        } catch (final Exception e) {
-            logNoDmaapEncoder(topic, e);
-            return Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(new Error(e.getMessage())).build();
+                    .entity(new Error(topic + " not acceptable due to current state"))
+                    .build();
+        } catch (Exception e) {
+            return Response.status(Response.Status.INTERNAL_SERVER_ERROR).entity(new Error(e.getMessage()))
+                            .build();
         }
     }
 
-    private void logNoDmaapEncoder(String topic, Exception ex) {
-        logger.debug("{}: cannot offer for encoder dmaap topic for {} because of {}", this, topic, ex.getMessage(), ex);
-    }
-
     /**
      * GET.
      *
@@ -2184,7 +2066,7 @@ public class RestManager {
         try {
             newLevel = LoggerUtil.setLevel(loggerName, loggerLevel);
         } catch (final IllegalArgumentException e) {
-            logger.warn("{}: no logger {}", this, loggerName, loggerLevel, e);
+            logger.warn("{}: invalid operation for logger {} and level {}", this, loggerName, loggerLevel, e);
             return Response.status(Status.NOT_FOUND).build();
         } catch (final IllegalStateException e) {
             logger.warn("{}: logging framework unavailable for {} / {}", this, loggerName, loggerLevel, e);
@@ -2223,9 +2105,7 @@ public class RestManager {
 
     @Override
     public String toString() {
-        final StringBuilder builder = new StringBuilder();
-        builder.append("rest-telemetry-api []");
-        return builder.toString();
+        return "rest-telemetry-api []";
     }
 
     /**
diff --git a/policy-management/src/test/java/org/onap/policy/drools/server/restful/test/RestManagerTest.java b/policy-management/src/test/java/org/onap/policy/drools/server/restful/test/RestManagerTest.java
index 11f27f49..c484497c 100644
--- a/policy-management/src/test/java/org/onap/policy/drools/server/restful/test/RestManagerTest.java
+++ b/policy-management/src/test/java/org/onap/policy/drools/server/restful/test/RestManagerTest.java
@@ -2,7 +2,7 @@
  * ============LICENSE_START=======================================================
  * policy-management
  * ================================================================================
- * Copyright (C) 2017-2018 AT&T Intellectual Property. All rights reserved.
+ * Copyright (C) 2017-2019 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.
@@ -92,7 +92,6 @@ public class RestManagerTest {
     private static final String DMAAP_SINK_PASSWD_KEY = PolicyEndPointProperties.PROPERTY_DMAAP_SINK_TOPICS + "."
             + DMAAP_TOPIC + PolicyEndPointProperties.PROPERTY_TOPIC_AAF_PASSWORD_SUFFIX;
 
-
     private static final String FOO_CONTROLLER_FILE = FOO_CONTROLLER + "-controller.properties";
     private static final String FOO_CONTROLLER_FILE_BAK = FOO_CONTROLLER_FILE + ".bak";
 
@@ -142,7 +141,6 @@ public class RestManagerTest {
         engineProps.put(DMAAP_SOURCE_PASSWD_KEY, DMAAP_PASSWD);
         engineProps.put(DMAAP_SINK_MECHID_KEY, DMAAP_MECHID);
         engineProps.put(DMAAP_SINK_PASSWD_KEY, DMAAP_PASSWD);
-        engineProps.put(PolicyEndPointProperties.PROPERTY_NOOP_SINK_TOPICS, NOOP_TOPIC);
 
         PolicyEngine.manager.configure(engineProps);
         PolicyEngine.manager.start();
@@ -160,6 +158,11 @@ public class RestManagerTest {
         if (!NetworkUtil.isTcpPortOpen("localhost", DEFAULT_TELEMETRY_PORT, 5, 10000L)) {
             throw new IllegalStateException("cannot connect to port " + DEFAULT_TELEMETRY_PORT);
         }
+
+        Properties noopProperties = new Properties();
+        noopProperties.put(PolicyEndPointProperties.PROPERTY_NOOP_SOURCE_TOPICS, NOOP_TOPIC);
+        noopProperties.put(PolicyEndPointProperties.PROPERTY_NOOP_SINK_TOPICS, NOOP_TOPIC);
+        TopicEndpoint.manager.addTopics(noopProperties);
     }
 
     /**
@@ -255,13 +258,25 @@ public class RestManagerTest {
         httpDelete.releaseConnection();
 
         /*
-         * PUT: /engine/topics/sources/ueb/topic/events /engine/topics/sources/dmaap/topic/events
-         * /engine/topics/switches/lock DELETE: /engine/topics/switches/lock
+         * PUT: /engine/topics/sources/ueb/topic/events
+         *      /engine/topics/sources/dmaap/topic/events
+         *      /engine/topics/switches/lock
+         *
+         * DELETE: /engine/topics/switches/lock
          */
         httpPut = new HttpPut(HOST_URL + "/engine/topics/sources/ueb/" + UEB_TOPIC + "/events");
         httpPut.addHeader("Content-Type", "text/plain");
         httpPut.addHeader("Accept", "application/json");
-        httpPut.setEntity(new StringEntity("FOOOO"));
+        httpPut.setEntity(new StringEntity("{x:y}"));
+        response = client.execute(httpPut);
+        logger.info(httpPut.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
+        assertEquals(200, response.getStatusLine().getStatusCode());
+        httpPut.releaseConnection();
+
+        httpPut = new HttpPut(HOST_URL + "/engine/topics/sources/noop/" + NOOP_TOPIC + "/events");
+        httpPut.addHeader("Content-Type", "text/plain");
+        httpPut.addHeader("Accept", "application/json");
+        httpPut.setEntity(new StringEntity("{x:y}"));
         response = client.execute(httpPut);
         logger.info(httpPut.getRequestLine() + " response code: {}", response.getStatusLine().getStatusCode());
         assertEquals(200, response.getStatusLine().getStatusCode());