Initial commit for WASB MVP

azure-storage/pom.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+		 xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+		 xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+	<modelVersion>4.0.0</modelVersion>
+
+	<artifactId>spring-cloud-stream-azure-storage-parent</artifactId>
+	<packaging>pom</packaging>
+
+    <properties>
+        <azure.version>4.3.0</azure.version>
+    </properties>
+
+	<parent>
+		<groupId>org.springframework.cloud.stream.app</groupId>
+		<artifactId>spring-cloud-stream-app-starters</artifactId>
+		<version>1.0.0.BUILD-SNAPSHOT</version>
+	</parent>
+
+	<modules>
+		<module>spring-cloud-starter-stream-sink-azure-blob</module>
+	</modules>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.microsoft.azure</groupId>
+            <artifactId>azure-storage</artifactId>
+            <version>${azure.version}</version>
+        </dependency>
+    </dependencies>
+
+</project>

azure-storage/spring-cloud-starter-stream-sink-azure-blob/README.md

@@ -0,0 +1,17 @@
+# spring-cloud-stream-wasb-sink
+#### A Windows Azure Storage Blob sink module for Spring Cloud Stream
+
+This module is an MVP implementation of a BlockBlob-based sink module
+for text payloads originating from a Spring Cloud Dataflow stream.
+
+An example stream definition:
+
+```
+dataflow:> app register --type sink --name wasb --uri file:///Users/kdunn/.m2/repository/io/pivotal/pde/wasb-sink/0.0.1-SNAPSHOT/wasb-sink-0.0.1-SNAPSHOT.jar
+
+# CloudBlockBlob (every payload overwrites)
+dataflow:> stream create --name testWasb --definition 'time | wasb --accountName="scdftest" --accountKey="<YOUR STORAGE ACCOUNT KEY HERE>" --containerName="time" --blobName="test" ' --deploy
+
+# CloudAppendBlob (every payload appends)
+dataflow:> stream create --name testWasb --definition 'time | wasb --accountName="scdftest" --accountKey="<YOUR STORAGE ACCOUNT KEY HERE>" --containerName="time" --blobName="testAppend" --appendOnly=true --overwriteExistingAppend=true ' --deploy
+```

azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml

@@ -0,0 +1,47 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>spring-cloud-starter-stream-sink-wasb</artifactId>
+    <name>spring-cloud-starter-stream-sink-wasb</name>
+    <description>Spring Cloud Stream Windows Azure Blob Storage Sink</description>
+
+    <parent>
+        <groupId>org.springframework.cloud.stream.app</groupId>
+        <artifactId>spring-cloud-stream-azure-storage-parent</artifactId>
+        <version>1.0.0.BUILD-SNAPSHOT</version>
+    </parent>
+
+    <properties>
+        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+        <java.version>1.8</java.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud</groupId>
+            <artifactId>spring-cloud-starter-stream-kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>3.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+    </dependencies>
+
+</project>

azure-storage/spring-cloud-starter-stream-sink-azure-blob/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/AzureBlobSinkConfiguration.java

@@ -0,0 +1,140 @@
+/*
+ * Copyright 2016 the original author or authors.
+ *
+ * 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.
+ */
+
+package org.springframework.cloud.stream.app.azure.storage.sink;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.cloud.stream.annotation.EnableBinding;
+import org.springframework.cloud.stream.messaging.Sink;
+import org.springframework.context.annotation.DependsOn;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.MessagingException;
+
+// Include the following imports to use blob APIs.
+import com.microsoft.azure.storage.CloudStorageAccount;
+import com.microsoft.azure.storage.blob.BlobContainerPermissions;
+import com.microsoft.azure.storage.blob.BlobContainerPublicAccessType;
+import com.microsoft.azure.storage.blob.CloudBlob;
+import com.microsoft.azure.storage.blob.CloudAppendBlob;
+import com.microsoft.azure.storage.blob.CloudBlobClient;
+import com.microsoft.azure.storage.blob.CloudBlobContainer;
+import com.microsoft.azure.storage.blob.CloudBlockBlob;
+
+
+/**
+ * @author Kyle Dunn
+ */
+@EnableBinding(Sink.class)
+@EnableConfigurationProperties(AzureBlobSinkProperties.class)
+@SpringBootApplication
+public class AzureBlobSinkConfiguration {
+
+    protected static Logger LOG = LoggerFactory.getLogger(AzureBlobSinkConfiguration.class);
+
+    @Autowired
+    private AzureBlobSinkProperties properties;
+
+    private CloudBlob blobService;
+
+    @Autowired
+    public void setBlobService() {
+        // Define the connection-string with your values
+        final String storageConnectionString =
+            "DefaultEndpointsProtocol=" + this.properties.getDefaultEndpointsProtocol() +
+            ";AccountName=" + this.properties.getAccountName() +
+            ";AccountKey=" + this.properties.getAccountKey();
+
+        try {
+            // Setup the cloud storage account.
+            CloudStorageAccount account = CloudStorageAccount.parse(storageConnectionString);
+
+            LOG.info("getBlobService() : using account {}", this.properties.getAccountName());
+
+            // Create a blob service client
+            CloudBlobClient blobClient = account.createCloudBlobClient();
+
+            // Get a reference to a container
+            // The container name must be lower case
+            CloudBlobContainer container = blobClient.getContainerReference(this.properties.getContainerName().toLowerCase());
+
+            LOG.info("getBlobService() : using container {}", this.properties.getContainerName());
+
+            if (this.properties.getAutoCreateContainer()) {
+                container.createIfNotExists();
+            }            
+
+            // Make the container public
+            if (this.properties.getPublicPermission()) {
+                LOG.info("getBlobService() : making container publically accessible");
+
+                // Create a permissions object
+                BlobContainerPermissions containerPermissions = new BlobContainerPermissions();
+
+                // Include public access in the permissions object
+                containerPermissions.setPublicAccess(BlobContainerPublicAccessType.CONTAINER);
+
+                // Set the permissions on the container
+                container.uploadPermissions(containerPermissions);
+            }
+
+            LOG.info("getBlobService() : using blob name {}", this.properties.getBlobName());
+
+            if (this.properties.getAppendOnly()) {
+                this.blobService = container.getAppendBlobReference(this.properties.getBlobName());
+                if (this.properties.getOverwiteExistingAppend()) {
+                    ((CloudAppendBlob) blobService).createOrReplace();
+                }
+            }
+            else {
+                this.blobService = container.getBlockBlobReference(this.properties.getBlobName());
+            }
+        } catch (Exception e) {
+            // Log the stack trace.
+            LOG.error("getBlobService() : {}", e.getMessage());
+            //throw e;
+        }
+    }
+
+    @ServiceActivator(inputChannel=Sink.INPUT)
+    public void pushToAzureBlob(Message<?> message) throws MessagingException {
+        try {
+            // Upload the payload to the blob
+            if (this.properties.getAppendOnly()) {
+                ((CloudAppendBlob) blobService).appendText(message.getPayload().toString() + "\n");
+            }
+            else {
+                ((CloudBlockBlob) blobService).uploadText(message.getPayload().toString());
+            }
+
+        } catch (Exception e) {
+            // Log the stack trace.
+            LOG.error("pushToAzureBlob() : {}", e.getMessage());
+        }
+
+    }
+
+    public static void main(String[] args) {
+        SpringApplication.run(AzureBlobSinkConfiguration.class, args);
+    }
+
+}