HTTP/webhook events

Author: jimmarino

extensions/transfer/transfer-demo-protocols/src/main/java/com/microsoft/dagx/transfer/demo/protocols/DemoProtocolsTransferExtension.java

@@ -3,17 +3,19 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.microsoft.dagx.spi.DagxSetting;
 import com.microsoft.dagx.spi.monitor.Monitor;
+import com.microsoft.dagx.spi.protocol.web.WebService;
 import com.microsoft.dagx.spi.security.Vault;
 import com.microsoft.dagx.spi.system.ServiceExtension;
 import com.microsoft.dagx.spi.system.ServiceExtensionContext;
 import com.microsoft.dagx.spi.transfer.flow.DataFlowManager;
 import com.microsoft.dagx.spi.transfer.provision.ProvisionManager;
 import com.microsoft.dagx.spi.transfer.provision.ResourceManifestGenerator;
+import com.microsoft.dagx.transfer.demo.protocols.http.PubSubHttpEndpoint;
 import com.microsoft.dagx.transfer.demo.protocols.object.DemoObjectStorage;
 import com.microsoft.dagx.transfer.demo.protocols.object.ObjectStorageFlowController;
 import com.microsoft.dagx.transfer.demo.protocols.spi.object.ObjectStorageMessage;
-import com.microsoft.dagx.transfer.demo.protocols.spi.stream.TopicManager;
 import com.microsoft.dagx.transfer.demo.protocols.spi.stream.StreamPublisherRegistry;
+import com.microsoft.dagx.transfer.demo.protocols.spi.stream.TopicManager;
 import com.microsoft.dagx.transfer.demo.protocols.spi.stream.message.ConnectMessage;
 import com.microsoft.dagx.transfer.demo.protocols.spi.stream.message.DataMessage;
 import com.microsoft.dagx.transfer.demo.protocols.spi.stream.message.PubSubMessage;
@@ -48,11 +50,15 @@
 public class DemoProtocolsTransferExtension implements ServiceExtension {
     @DagxSetting
     private static final String WS_PUBSUB_ENDPOINT = "dagx.demo.protocol.ws.pubsub";
-
     private static final String DEFAULT_WS_PUBSUB_ENDPOINT = "ws://localhost:8181/pubsub/";
 
+    @DagxSetting
+    private static final String HTTP_PUBSUB_ENDPOINT = "dagx.demo.protocol.http.pubsub";
+    private static final String DEFAULT_HTTP_PUBSUB_ENDPOINT = "http://localhost:8181/api/demo/pubsub/";
+
     DemoObjectStorage objectStorage;
     DemoTopicManager topicManager;
+    PubSubHttpEndpoint httpEndpoint;
 
     private Monitor monitor;
 
@@ -78,6 +84,9 @@ public void initialize(ServiceExtensionContext context) {
         var jettyService = context.getService(JettyService.class);
         new WebSocketFactory().publishEndpoint(PubSubServerEndpoint.class, () -> new PubSubServerEndpoint(topicManager, objectMapper, monitor), jettyService);
 
+        httpEndpoint = new PubSubHttpEndpoint(topicManager);
+        context.getService(WebService.class).registerController(httpEndpoint);
+
         registerGenerators(context);
 
         registerProvisioners(topicManager, context);
@@ -91,6 +100,7 @@ public void initialize(ServiceExtensionContext context) {
     public void start() {
         objectStorage.start();
         topicManager.start();
+        httpEndpoint.start();
     }
 
     @Override
@@ -101,13 +111,17 @@ public void shutdown() {
         if (topicManager != null) {
             topicManager.stop();
         }
+        if (httpEndpoint != null) {
+            httpEndpoint.stop();
+        }
     }
 
     private void registerGenerators(ServiceExtensionContext context) {
         var manifestGenerator = context.getService(ResourceManifestGenerator.class);
 
-        var endpointAddress = context.getSetting(WS_PUBSUB_ENDPOINT, DEFAULT_WS_PUBSUB_ENDPOINT);
-        manifestGenerator.registerClientGenerator(new PushStreamResourceGenerator(endpointAddress));
+        var wsEndpointAddress = context.getSetting(WS_PUBSUB_ENDPOINT, DEFAULT_WS_PUBSUB_ENDPOINT);
+        var httpEndpointAddress = context.getSetting(HTTP_PUBSUB_ENDPOINT, DEFAULT_HTTP_PUBSUB_ENDPOINT);
+        manifestGenerator.registerClientGenerator(new PushStreamResourceGenerator(wsEndpointAddress, httpEndpointAddress));
     }
 
     private void registerProvisioners(TopicManager topicManager, ServiceExtensionContext context) {

extensions/transfer/transfer-demo-protocols/src/main/java/com/microsoft/dagx/transfer/demo/protocols/http/HttpStreamSession.java

@@ -1,25 +1,49 @@
 package com.microsoft.dagx.transfer.demo.protocols.http;
 
+import com.microsoft.dagx.spi.DagxException;
 import com.microsoft.dagx.transfer.demo.protocols.stream.StreamSession;
+import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
 
+import java.io.IOException;
 import java.net.URL;
 
 /**
  * Publishes to an HTTP endpoint.
  */
 public class HttpStreamSession implements StreamSession {
     private URL endpointURL;
+    private String destinationToken;
     private OkHttpClient httpClient;
 
-    public HttpStreamSession(URL endpointURL, OkHttpClient httpClient) {
+    public HttpStreamSession(URL endpointURL, String destinationToken, OkHttpClient httpClient) {
         this.endpointURL = endpointURL;
+        this.destinationToken = destinationToken;
         this.httpClient = httpClient;
     }
 
     @Override
     public void publish(byte[] data) {
-        throw new UnsupportedOperationException();
+        try {
+            var body = RequestBody.create(data, MediaType.get("application/json"));
+            Request request = new Request.Builder()
+                    .url(endpointURL)
+                    .addHeader("Content-Type", "application/json")
+                    .addHeader("X-Authorization", destinationToken)
+                    .post(body)
+                    .build();
+
+            try (var response = httpClient.newCall(request).execute()) {
+                if (response.code() != 200) {
+                    throw new DagxException("Invalid response received from destination: " + response.code());
+                }
+            }
+
+        } catch (IOException e) {
+            throw new DagxException(e);
+        }
     }
 
     @Override

extensions/transfer/transfer-demo-protocols/src/main/java/com/microsoft/dagx/transfer/demo/protocols/http/PubSubHttpEndpoint.java

@@ -1,24 +1,88 @@
 package com.microsoft.dagx.transfer.demo.protocols.http;
 
+import com.microsoft.dagx.spi.DagxException;
+import com.microsoft.dagx.transfer.demo.protocols.spi.stream.TopicManager;
 import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.HeaderParam;
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.PathParam;
 import jakarta.ws.rs.Produces;
 import jakarta.ws.rs.core.MediaType;
+import jakarta.ws.rs.core.Response;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
 
 /**
- *
+ * A JAX-RS endpoint that accepts data to be published to a topic.
  */
 @Consumes({MediaType.APPLICATION_JSON})
 @Produces({MediaType.APPLICATION_JSON})
 @Path("/demo/pubsub")
 public class PubSubHttpEndpoint {
+    private TopicManager topicManager;
+
+    private AtomicBoolean active = new AtomicBoolean();
+
+    private ExecutorService executorService;
+    private LinkedBlockingQueue<QueueEntry> queue = new LinkedBlockingQueue<>();
+
+    public PubSubHttpEndpoint(TopicManager topicManager) {
+        this.topicManager = topicManager;
+        executorService = Executors.newSingleThreadExecutor();
+    }
+
+    public void start() {
+        active.set(true);
+        executorService.submit(this::run);
+    }
+
+    public void stop() {
+        active.set(false);
+        if (executorService != null) {
+            executorService.shutdownNow();
+        }
+    }
 
     @POST
     @Path("{destinationName}")
-    public void publish(@PathParam("destinationName") String destinationName) {
+    public Response publish(@PathParam("destinationName") String topicName, @HeaderParam("X-Authorization") String token, byte[] data) {
+        var result = topicManager.connect(topicName, token);
+        if (result.success()) {
+            var entry = new QueueEntry(result.getConsumer(), data);
+            try {
+                queue.put(entry);
+            } catch (InterruptedException e) {
+                Thread.interrupted();
+                throw new DagxException(e);
+            }
+            return Response.ok().build();
+        } else {
+            return Response.status(Response.Status.BAD_REQUEST).build();
+        }
+    }
+
+    private static class QueueEntry {
+        Consumer<byte[]> connection;
+        byte[] data;
+
+        public QueueEntry(Consumer<byte[]> connection, byte[] data) {
+            this.connection = connection;
+            this.data = data;
+        }
+    }
 
+    private void run() {
+        while (active.get()) {
+            var entry = queue.poll();
+            if (entry != null) {
+                entry.connection.accept(entry.data);
+            }
+        }
     }
 
 }

extensions/transfer/transfer-demo-protocols/src/main/java/com/microsoft/dagx/transfer/demo/protocols/spi/DemoProtocols.java

@@ -9,7 +9,9 @@ public interface DemoProtocols {
 
     String PULL_STREAM = "dagx:demo:pullstream";
 
-    String PUSH_STREAM = "dagx:demo:pushstream";
+    String PUSH_STREAM_WS = "dagx:demo:pushstream:ws";
+
+    String PUSH_STREAM_HTTP = "dagx:demo:pushstream:http";
 
     String ENDPOINT_ADDRESS = "endpointAddress";
 

extensions/transfer/transfer-demo-protocols/src/main/java/com/microsoft/dagx/transfer/demo/protocols/stream/PushStreamContext.java

@@ -37,14 +37,15 @@ public StreamSession createSession(String uri, String topicName, String secretNa
         var accessToken = vault.resolveSecret(secretName);
         var destinationToken = readAccessToken(accessToken, secretName);
         var endpointUri = URI.create(uri);
+
         if ("ws".equalsIgnoreCase(endpointUri.getScheme())) {
             var session = new WsPushStreamSession(endpointUri, topicName, destinationToken, objectMapper, monitor);
             session.connect();
             return session;
         } else if ("https".equalsIgnoreCase(endpointUri.getScheme()) || "http".equalsIgnoreCase(endpointUri.getScheme())) {
             try {
-                var endpointUrl = endpointUri.toURL();
-                return new HttpStreamSession(endpointUrl, httpClient);
+                var endpointUrl = endpointUri.resolve(topicName).toURL();
+                return new HttpStreamSession(endpointUrl, destinationToken, httpClient);
             } catch (MalformedURLException e) {
                 throw new DagxException(e);
             }

extensions/transfer/transfer-demo-protocols/src/main/java/com/microsoft/dagx/transfer/demo/protocols/stream/PushStreamFlowController.java

@@ -19,7 +19,7 @@ public PushStreamFlowController(StreamPublisherRegistry registry) {
 
     @Override
     public boolean canHandle(DataRequest dataRequest) {
-        return DemoProtocols.PUSH_STREAM.equals(dataRequest.getDestinationType());
+        return DemoProtocols.PUSH_STREAM_WS.equals(dataRequest.getDestinationType()) ||  DemoProtocols.PUSH_STREAM_HTTP.equals(dataRequest.getDestinationType()) ;
     }
 
     @Override

extensions/transfer/transfer-demo-protocols/src/main/java/com/microsoft/dagx/transfer/demo/protocols/stream/PushStreamProvisionedResourceDefinition.java

@@ -36,7 +36,7 @@ public String getResourceName() {
     @Override
     public DataAddress createDataDestination() {
         return DataAddress.Builder.newInstance()
-                .type(DemoProtocols.PUSH_STREAM)
+                .type(DemoProtocols.PUSH_STREAM_WS)
                 .keyName("demo-temp-" + destinationName)
                 .property(DemoProtocols.ENDPOINT_ADDRESS, endpointAddress)
                 .property(DemoProtocols.DESTINATION_NAME, destinationName)

extensions/transfer/transfer-demo-protocols/src/main/java/com/microsoft/dagx/transfer/demo/protocols/stream/PushStreamResourceGenerator.java

@@ -12,24 +12,30 @@
  * generated. Otherwise, a definition containing metadata to create a destination topic will be returned.
  */
 public class PushStreamResourceGenerator implements ResourceDefinitionGenerator {
-    private final String endpointAddress;
+    private final String wsEndpointAddress;
+    private final String httpEndpointAddress;
 
-    public PushStreamResourceGenerator(String endpointAddress) {
-        this.endpointAddress = endpointAddress;
+    public PushStreamResourceGenerator(String wsEndpointAddress, String httpEndpointAddress) {
+        this.wsEndpointAddress = wsEndpointAddress;
+        this.httpEndpointAddress = httpEndpointAddress;
     }
 
     @Override
     public @Nullable PushStreamResourceDefinition generate(TransferProcess process) {
         var dataRequest = process.getDataRequest();
-
-        if (!DemoProtocols.PUSH_STREAM.equals(dataRequest.getDestinationType())) {
-            return null;
-        }
         if (!dataRequest.isManagedResources()) {
             // The resource is unmanaged, which means it was created by an external system. In this case it does not need to be provisioned.
             return null;
         }
 
+        String endpointAddress;
+        if (DemoProtocols.PUSH_STREAM_WS.equals(dataRequest.getDestinationType())) {
+            endpointAddress = wsEndpointAddress;
+        } else if (DemoProtocols.PUSH_STREAM_HTTP.equals(dataRequest.getDestinationType())) {
+            endpointAddress = httpEndpointAddress;
+        } else {
+            return null;
+        }
         var destinationName = dataRequest.getDataDestination().getProperty(DemoProtocols.DESTINATION_NAME);
         if (destinationName == null) {
             destinationName = UUID.randomUUID().toString();