diff --git a/azure-storage/pom.xml b/azure-storage/pom.xml new file mode 100644 index 000000000..73b59e6ea --- /dev/null +++ b/azure-storage/pom.xml @@ -0,0 +1,32 @@ + + + 4.0.0 + + spring-cloud-stream-azure-storage-parent + pom + + + 4.3.0 + + + + org.springframework.cloud.stream.app + spring-cloud-stream-app-starters + 1.0.0.BUILD-SNAPSHOT + + + + spring-cloud-starter-stream-sink-azure-blob + + + + + com.microsoft.azure + azure-storage + ${azure.version} + + + + diff --git a/azure-storage/spring-cloud-starter-stream-sink-azure-blob/README.md b/azure-storage/spring-cloud-starter-stream-sink-azure-blob/README.md new file mode 100644 index 000000000..dd329e49d --- /dev/null +++ b/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="" --containerName="time" --blobName="test" ' --deploy + +# CloudAppendBlob (every payload appends) +dataflow:> stream create --name testWasb --definition 'time | wasb --accountName="scdftest" --accountKey="" --containerName="time" --blobName="testAppend" --appendOnly=true --overwriteExistingAppend=true ' --deploy +``` diff --git a/azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml b/azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml new file mode 100644 index 000000000..7d2330552 --- /dev/null +++ b/azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml @@ -0,0 +1,28 @@ + + + 4.0.0 + + spring-cloud-starter-stream-sink-wasb + spring-cloud-starter-stream-sink-wasb + Spring Cloud Stream Windows Azure Blob Storage Sink + + + org.springframework.cloud.stream.app + spring-cloud-stream-azure-storage-parent + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + + + + + org.springframework.cloud.stream.app + app-starters-test-support + test + + + + diff --git a/azure-storage/spring-cloud-starter-stream-sink-azure-blob/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/AzureBlobSinkConfiguration.java b/azure-storage/spring-cloud-starter-stream-sink-azure-blob/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/AzureBlobSinkConfiguration.java new file mode 100644 index 000000000..05e92878b --- /dev/null +++ b/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); + } + +} diff --git a/azure-storage/spring-cloud-starter-stream-sink-azure-blob/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/AzureBlobSinkProperties.java b/azure-storage/spring-cloud-starter-stream-sink-azure-blob/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/AzureBlobSinkProperties.java new file mode 100644 index 000000000..c0adbe268 --- /dev/null +++ b/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; + } +} diff --git a/pom.xml b/pom.xml index ff33ec311..cbc72b859 100644 --- a/pom.xml +++ b/pom.xml @@ -32,6 +32,7 @@ app-starters-test-support aws-integration-tests aws-s3 + azure-storage cassandra cloudfoundry file diff --git a/spring-cloud-stream-app-generator/pom.xml b/spring-cloud-stream-app-generator/pom.xml index 9e2346d41..0221d19d9 100644 --- a/spring-cloud-stream-app-generator/pom.xml +++ b/spring-cloud-stream-app-generator/pom.xml @@ -43,6 +43,10 @@ org.springframework.cloud.stream.app.test.aggregate.counter.AggregateCounterSinkTestConfiguration.class + + spring.cloud.starter.stream.sink.azure.blob.AzureBlobSinkConfiguration.class + true + true