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,28 @@
+<?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>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework.cloud.stream.app</groupId>
+            <artifactId>app-starters-test-support</artifactId>
+            <scope>test</scope>
+        </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,132 @@
+/*
+ * 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 java.io.IOException;
+import java.net.URISyntaxException;
+import java.security.InvalidKeyException;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+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.StorageException;
+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 {
+
+    @Autowired
+    private AzureBlobSinkProperties properties;
+
+    private static Log logger = LogFactory.getLog(AzureBlobSinkConfiguration.class);
+
+    private CloudBlob blobService;
+
+    @Autowired
+    public void setBlobService() throws StorageException, URISyntaxException, InvalidKeyException {
+        // Define the connection-string with your values
+        final String storageConnectionString =
+            "DefaultEndpointsProtocol=" + this.properties.getDefaultEndpointsProtocol() +
+            ";AccountName=" + this.properties.getAccountName() +
+            ";AccountKey=" + this.properties.getAccountKey();
+
+        // Setup the cloud storage account.
+        CloudStorageAccount account = CloudStorageAccount.parse(storageConnectionString);
+
+        logger.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());
+
+        logger.info("getBlobService() : using container " + this.properties.getContainerName());
+
+        if (this.properties.getAutoCreateContainer()) {
+            container.createIfNotExists();
+        }            
+
+        // Make the container public
+        if (this.properties.getPublicPermission()) {
+            logger.info("getBlobService() : making container publicly 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);
+        }
+
+        logger.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());
+        }
+    }
+
+    @ServiceActivator(inputChannel=Sink.INPUT)
+    public void pushToAzureBlob(Message<?> message) throws StorageException, IOException {
+        // Upload the payload to the blob
+        if (this.properties.getAppendOnly()) {
+            ((CloudAppendBlob) blobService).appendText(message.getPayload().toString());
+        }
+        else {
+            ((CloudBlockBlob) blobService).uploadText(message.getPayload().toString());
+        }
+    }
+
+    public static void main(String[] args) {
+        SpringApplication.run(AzureBlobSinkConfiguration.class, args);
+    }
+
+}

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

@@ -0,0 +1,155 @@
+/*
+ * 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 static org.springframework.integration.handler.LoggingHandler.Level.*;
+
+import javax.validation.constraints.NotNull;
+
+import org.hibernate.validator.constraints.NotBlank;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.integration.handler.LoggingHandler;
+
+/**
+ * Configuration properties for the Azure Blob Sink module.
+ *
+ * @author Kyle Dunn
+ */
+@ConfigurationProperties("azure.blob")
+public class AzureBlobSinkProperties {
+
+    /**
+     * The default Azure endpoint protocol.
+     */
+    private String defaultEndpointsProtocol = "http";
+
+    /**
+     * The Azure Storage Account name.
+     */
+    private String account;
+
+    /**
+     * The Azure Storage Account key.
+     */
+    private String key;
+
+    /**
+     * The Azure Storage Container name.
+     */
+    private String container;
+
+    /**
+     * The Azure Storage Blob name.
+     */
+    private String blob;
+
+    /**
+     * Create a container if it doesn't already exist.
+     */
+    private boolean createContainer = true;
+
+    /**
+     * Set container access policy to public.
+     */
+    private boolean publicPermission = true;
+
+    /**
+     * Specify if using an CloudAppendBlob
+     */
+    private boolean appendOnly = true;
+
+    /**
+     * Specify whether to silently overwrite 
+     * existing an CloudAppendBlob with the same name
+     */
+    private boolean overwriteExistingAppend = false;
+
+    public String getDefaultEndpointsProtocol() {
+        return defaultEndpointsProtocol;
+    }
+
+    public void setDefaultEndpointsProtocol(String p) {
+        this.defaultEndpointsProtocol = p;
+    }
+
+    public String getAccountName() {
+        return account;
+    }
+
+    public void setAccountName(String n) {
+        this.account = n;
+    }
+
+    public String getAccountKey() {
+        return key;
+    }
+
+    public void setAccountKey(String k) {
+        this.key = k;
+    }
+
+    public String getContainerName() {
+        return container;
+    }
+
+    public void setContainerName(String c) {
+        this.container = c;
+    }
+
+    public String getBlobName() {
+        return blob;
+    }
+
+    public void setBlobName(String b) {
+        this.blob = b;
+    }
+
+    public boolean getAutoCreateContainer() {
+        return createContainer;
+    }
+
+    // Automatically create the container
+    public void setAutoCreateContainer(Boolean b) {
+        this.createContainer = b;
+    }
+
+    public boolean getPublicPermission() {
+        return publicPermission;
+    }
+
+    // Include public access in the permissions object 
+    public void setPublicPermission(Boolean b) {
+        this.publicPermission = b;
+    }
+
+    public boolean getAppendOnly() {
+        return appendOnly;
+    }
+
+    public void setAppendOnly(Boolean appendOnly) {
+        this.appendOnly = appendOnly;
+    }
+
+    public boolean getOverwiteExistingAppend() {
+        return overwriteExistingAppend;
+    }
+
+    public void setOverwriteExistingAppend(Boolean overwrite) {
+        this.overwriteExistingAppend = overwrite;
+    }
+}

pom.xml

@@ -32,6 +32,7 @@
 		<module>app-starters-test-support</module>
 		<module>aws-integration-tests</module>
 		<module>aws-s3</module>
+		<module>azure-storage</module>
 		<module>cassandra</module>
 		<module>cloudfoundry</module>
 		<module>file</module>

spring-cloud-stream-app-generator/pom.xml

@@ -43,6 +43,10 @@
 						<aggregate-counter-sink>
 							<extraTestConfigClass>org.springframework.cloud.stream.app.test.aggregate.counter.AggregateCounterSinkTestConfiguration.class</extraTestConfigClass>
 						</aggregate-counter-sink>
+						<azure-blob-sink>
+							<autoConfigClass>spring.cloud.starter.stream.sink.azure.blob.AzureBlobSinkConfiguration.class</autoConfigClass>
+							<testsIgnored>true</testsIgnored>
+						</azure-blob-sink>
 						<bridge-processor />
 						<cassandra-sink>
 							<testsIgnored>true</testsIgnored>