From 3bfa5729c76f88267e8c523055a464342808e46e Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Wed, 13 Jul 2016 17:44:51 -0600 Subject: [PATCH 01/17] Initial commit for WASB MVP --- azure-storage/pom.xml | 20 ++ .../README.md | 17 ++ .../pom.xml | 83 ++++++++ .../storage/sink/WasbSinkConfiguration.java | 141 ++++++++++++++ .../storage/sink/WasbSinkProperties.java | 184 ++++++++++++++++++ 5 files changed, 445 insertions(+) create mode 100644 azure-storage/pom.xml create mode 100644 azure-storage/spring-cloud-starter-stream-sink-wasb/README.md create mode 100644 azure-storage/spring-cloud-starter-stream-sink-wasb/pom.xml create mode 100644 azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkConfiguration.java create mode 100644 azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkProperties.java diff --git a/azure-storage/pom.xml b/azure-storage/pom.xml new file mode 100644 index 000000000..08c9ae6a2 --- /dev/null +++ b/azure-storage/pom.xml @@ -0,0 +1,20 @@ + + + 4.0.0 + + spring-cloud-stream-azure-storage-parent + pom + + + org.springframework.cloud.stream.app + spring-cloud-stream-app-starters + 1.0.0.BUILD-SNAPSHOT + + + + spring-cloud-starter-stream-sink-wasb + + + diff --git a/azure-storage/spring-cloud-starter-stream-sink-wasb/README.md b/azure-storage/spring-cloud-starter-stream-sink-wasb/README.md new file mode 100644 index 000000000..dd329e49d --- /dev/null +++ b/azure-storage/spring-cloud-starter-stream-sink-wasb/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-wasb/pom.xml b/azure-storage/spring-cloud-starter-stream-sink-wasb/pom.xml new file mode 100644 index 000000000..714525a75 --- /dev/null +++ b/azure-storage/spring-cloud-starter-stream-sink-wasb/pom.xml @@ -0,0 +1,83 @@ + + + 4.0.0 + + spring-cloud-starter-stream-sink-wasb + spring-cloud-starter-stream-sink-wasb + Spring Cloud Stream Windows Azure Storage Blob (WASB) Sink + + + org.springframework.cloud.stream.app + spring-cloud-stream-azure-storage-parent + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + 1.8 + 4.3.0 + + + + + maven2-repository.dev.java.net + Java.net repository + http://download.java.net/maven/2 + + + + + + org.springframework.cloud + spring-cloud-starter-stream-kafka + + + org.springframework.boot + spring-boot-starter-test + test + + + org.apache.commons + commons-lang3 + 3.0 + + + org.slf4j + slf4j-api + + + org.springframework.boot + spring-boot-configuration-processor + true + + + com.microsoft.azure + azure-storage + ${azure.version} + + + + + + + org.springframework.cloud + spring-cloud-dependencies + Brixton.SR1 + pom + import + + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + + diff --git a/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkConfiguration.java b/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkConfiguration.java new file mode 100644 index 000000000..ca814dde7 --- /dev/null +++ b/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkConfiguration.java @@ -0,0 +1,141 @@ +/* + * 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(WasbSinkProperties.class) +@SpringBootApplication +public class WasbSinkConfiguration { + + protected static Logger LOG = LoggerFactory.getLogger(WasbSinkConfiguration.class); + + @Autowired + private WasbSinkProperties properties; + + private CloudBlob blobService; + + @Autowired + @DependsOn("WasbSinkProperties") + 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 pushToWasb(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("pushToWasb() : {}", e.getMessage()); + } + + } + + public static void main(String[] args) { + SpringApplication.run(WasbSinkConfiguration.class, args); + } + +} diff --git a/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkProperties.java b/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkProperties.java new file mode 100644 index 000000000..d08662968 --- /dev/null +++ b/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkProperties.java @@ -0,0 +1,184 @@ +/* + * 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 Wasb Sink module. + * + * @author Kyle Dunn + */ +@ConfigurationProperties +public class WasbSinkProperties { + + /** + * The name to use. + */ + @Value("${spring.application.name:wasb.sink}") + private String name; + + /** + * The level at which to log messages. + */ + private LoggingHandler.Level level = INFO; + + /** + * The default Azure endpoint protocol. + */ + private String defaultEndpointsProtocol = "http"; + + /** + * The Azure Storage Account name. + */ + private String accountName; + + /** + * The Azure Storage Account key. + */ + private String accountKey; + + /** + * The Azure Storage Container name. + */ + private String containerName; + + /** + * The Azure Storage Blob name. + */ + private String blobName; + + /** + * Set container access policy to public. + */ + private Boolean autoCreateContainer = true; + + /** + * Create a container if it doesn't already exist. + */ + 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; + + @NotBlank + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + @NotNull + public LoggingHandler.Level getLevel() { + return level; + } + + public void setLevel(LoggingHandler.Level level) { + this.level = level; + } + + public String getDefaultEndpointsProtocol() { + return defaultEndpointsProtocol; + } + + public void setDefaultEndpointsProtocol(String p) { + this.defaultEndpointsProtocol = p; + } + + public String getAccountName() { + return accountName; + } + + public void setAccountName(String n) { + this.accountName = n; + } + + public String getAccountKey() { + return accountKey; + } + + public void setAccountKey(String k) { + this.accountKey = k; + } + + public String getContainerName() { + return containerName; + } + + public void setContainerName(String c) { + this.containerName = c; + } + + public String getBlobName() { + return blobName; + } + + public void setBlobName(String b) { + this.blobName = b; + } + + public Boolean getAutoCreateContainer() { + return autoCreateContainer; + } + + // Automatically create the container + public void setAutoCreateContainer(Boolean b) { + this.autoCreateContainer = 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; + } +} From 4a3dfb39fa8a2fa71333bc0a516acde1b426d517 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 10:33:26 -0600 Subject: [PATCH 02/17] Implementing some convention changes from the PR --- azure-storage/pom.xml | 14 +- .../README.md | 17 -- .../pom.xml | 83 -------- .../storage/sink/WasbSinkConfiguration.java | 141 -------------- .../storage/sink/WasbSinkProperties.java | 184 ------------------ 5 files changed, 13 insertions(+), 426 deletions(-) delete mode 100644 azure-storage/spring-cloud-starter-stream-sink-wasb/README.md delete mode 100644 azure-storage/spring-cloud-starter-stream-sink-wasb/pom.xml delete mode 100644 azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkConfiguration.java delete mode 100644 azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkProperties.java diff --git a/azure-storage/pom.xml b/azure-storage/pom.xml index 08c9ae6a2..73b59e6ea 100644 --- a/azure-storage/pom.xml +++ b/azure-storage/pom.xml @@ -7,6 +7,10 @@ spring-cloud-stream-azure-storage-parent pom + + 4.3.0 + + org.springframework.cloud.stream.app spring-cloud-stream-app-starters @@ -14,7 +18,15 @@ - spring-cloud-starter-stream-sink-wasb + spring-cloud-starter-stream-sink-azure-blob + + + com.microsoft.azure + azure-storage + ${azure.version} + + + diff --git a/azure-storage/spring-cloud-starter-stream-sink-wasb/README.md b/azure-storage/spring-cloud-starter-stream-sink-wasb/README.md deleted file mode 100644 index dd329e49d..000000000 --- a/azure-storage/spring-cloud-starter-stream-sink-wasb/README.md +++ /dev/null @@ -1,17 +0,0 @@ -# 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-wasb/pom.xml b/azure-storage/spring-cloud-starter-stream-sink-wasb/pom.xml deleted file mode 100644 index 714525a75..000000000 --- a/azure-storage/spring-cloud-starter-stream-sink-wasb/pom.xml +++ /dev/null @@ -1,83 +0,0 @@ - - - 4.0.0 - - spring-cloud-starter-stream-sink-wasb - spring-cloud-starter-stream-sink-wasb - Spring Cloud Stream Windows Azure Storage Blob (WASB) Sink - - - org.springframework.cloud.stream.app - spring-cloud-stream-azure-storage-parent - 1.0.0.BUILD-SNAPSHOT - - - - UTF-8 - 1.8 - 4.3.0 - - - - - maven2-repository.dev.java.net - Java.net repository - http://download.java.net/maven/2 - - - - - - org.springframework.cloud - spring-cloud-starter-stream-kafka - - - org.springframework.boot - spring-boot-starter-test - test - - - org.apache.commons - commons-lang3 - 3.0 - - - org.slf4j - slf4j-api - - - org.springframework.boot - spring-boot-configuration-processor - true - - - com.microsoft.azure - azure-storage - ${azure.version} - - - - - - - org.springframework.cloud - spring-cloud-dependencies - Brixton.SR1 - pom - import - - - - - - - - org.springframework.boot - spring-boot-maven-plugin - - - - - - diff --git a/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkConfiguration.java b/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkConfiguration.java deleted file mode 100644 index ca814dde7..000000000 --- a/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkConfiguration.java +++ /dev/null @@ -1,141 +0,0 @@ -/* - * 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(WasbSinkProperties.class) -@SpringBootApplication -public class WasbSinkConfiguration { - - protected static Logger LOG = LoggerFactory.getLogger(WasbSinkConfiguration.class); - - @Autowired - private WasbSinkProperties properties; - - private CloudBlob blobService; - - @Autowired - @DependsOn("WasbSinkProperties") - 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 pushToWasb(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("pushToWasb() : {}", e.getMessage()); - } - - } - - public static void main(String[] args) { - SpringApplication.run(WasbSinkConfiguration.class, args); - } - -} diff --git a/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkProperties.java b/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkProperties.java deleted file mode 100644 index d08662968..000000000 --- a/azure-storage/spring-cloud-starter-stream-sink-wasb/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/WasbSinkProperties.java +++ /dev/null @@ -1,184 +0,0 @@ -/* - * 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 Wasb Sink module. - * - * @author Kyle Dunn - */ -@ConfigurationProperties -public class WasbSinkProperties { - - /** - * The name to use. - */ - @Value("${spring.application.name:wasb.sink}") - private String name; - - /** - * The level at which to log messages. - */ - private LoggingHandler.Level level = INFO; - - /** - * The default Azure endpoint protocol. - */ - private String defaultEndpointsProtocol = "http"; - - /** - * The Azure Storage Account name. - */ - private String accountName; - - /** - * The Azure Storage Account key. - */ - private String accountKey; - - /** - * The Azure Storage Container name. - */ - private String containerName; - - /** - * The Azure Storage Blob name. - */ - private String blobName; - - /** - * Set container access policy to public. - */ - private Boolean autoCreateContainer = true; - - /** - * Create a container if it doesn't already exist. - */ - 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; - - @NotBlank - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - @NotNull - public LoggingHandler.Level getLevel() { - return level; - } - - public void setLevel(LoggingHandler.Level level) { - this.level = level; - } - - public String getDefaultEndpointsProtocol() { - return defaultEndpointsProtocol; - } - - public void setDefaultEndpointsProtocol(String p) { - this.defaultEndpointsProtocol = p; - } - - public String getAccountName() { - return accountName; - } - - public void setAccountName(String n) { - this.accountName = n; - } - - public String getAccountKey() { - return accountKey; - } - - public void setAccountKey(String k) { - this.accountKey = k; - } - - public String getContainerName() { - return containerName; - } - - public void setContainerName(String c) { - this.containerName = c; - } - - public String getBlobName() { - return blobName; - } - - public void setBlobName(String b) { - this.blobName = b; - } - - public Boolean getAutoCreateContainer() { - return autoCreateContainer; - } - - // Automatically create the container - public void setAutoCreateContainer(Boolean b) { - this.autoCreateContainer = 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; - } -} From 8139bdd2e641d384ad1df31a7870ed28a9348cb3 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 10:41:02 -0600 Subject: [PATCH 03/17] Adding implementation back in with new name --- .../README.md | 17 ++ .../pom.xml | 70 +++++++ .../sink/AzureBlobSinkConfiguration.java | 140 +++++++++++++ .../storage/sink/AzureBlobSinkProperties.java | 184 ++++++++++++++++++ 4 files changed, 411 insertions(+) create mode 100644 azure-storage/spring-cloud-starter-stream-sink-azure-blob/README.md create mode 100644 azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml create mode 100644 azure-storage/spring-cloud-starter-stream-sink-azure-blob/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/AzureBlobSinkConfiguration.java create mode 100644 azure-storage/spring-cloud-starter-stream-sink-azure-blob/src/main/java/org/springframework/cloud/stream/app/azure/storage/sink/AzureBlobSinkProperties.java 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..af240b9df --- /dev/null +++ b/azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml @@ -0,0 +1,70 @@ + + + 4.0.0 + + spring-cloud-starter-stream-sink-wasb + spring-cloud-starter-stream-sink-wasb + Spring Cloud Stream Windows Azure Storage Blob (WASB) Sink + + + org.springframework.cloud.stream.app + spring-cloud-stream-azure-storage-parent + 1.0.0.BUILD-SNAPSHOT + + + + UTF-8 + 1.8 + 4.3.0 + + + + + org.springframework.cloud + spring-cloud-starter-stream-kafka + + + org.springframework.boot + spring-boot-starter-test + test + + + org.apache.commons + commons-lang3 + 3.0 + + + org.slf4j + slf4j-api + + + org.springframework.boot + spring-boot-configuration-processor + true + + + + + + + org.springframework.cloud + spring-cloud-dependencies + Brixton.SR1 + pom + import + + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + + 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..02483709a --- /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,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); + } + +} 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..d0cf56427 --- /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,184 @@ +/* + * 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-sink") +public class AzureBlobSinkProperties { + + /** + * The name to use. + */ + @Value("${spring.application.name:azure-blob.sink}") + private String name; + + /** + * The level at which to log messages. + */ + private LoggingHandler.Level level = INFO; + + /** + * The default Azure endpoint protocol. + */ + private String defaultEndpointsProtocol = "http"; + + /** + * The Azure Storage Account name. + */ + private String accountName; + + /** + * The Azure Storage Account key. + */ + private String accountKey; + + /** + * The Azure Storage Container name. + */ + private String containerName; + + /** + * The Azure Storage Blob name. + */ + private String blobName; + + /** + * Set container access policy to public. + */ + private boolean autoCreateContainer = true; + + /** + * Create a container if it doesn't already exist. + */ + 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; + + @NotBlank + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } + + @NotNull + public LoggingHandler.Level getLevel() { + return level; + } + + public void setLevel(LoggingHandler.Level level) { + this.level = level; + } + + public String getDefaultEndpointsProtocol() { + return defaultEndpointsProtocol; + } + + public void setDefaultEndpointsProtocol(String p) { + this.defaultEndpointsProtocol = p; + } + + public String getAccountName() { + return accountName; + } + + public void setAccountName(String n) { + this.accountName = n; + } + + public String getAccountKey() { + return accountKey; + } + + public void setAccountKey(String k) { + this.accountKey = k; + } + + public String getContainerName() { + return containerName; + } + + public void setContainerName(String c) { + this.containerName = c; + } + + public String getBlobName() { + return blobName; + } + + public void setBlobName(String b) { + this.blobName = b; + } + + public boolean getAutoCreateContainer() { + return autoCreateContainer; + } + + // Automatically create the container + public void setAutoCreateContainer(Boolean b) { + this.autoCreateContainer = 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; + } +} From 0dd13470f36ca4b3d41690a134a3c7477757e600 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 10:47:06 -0600 Subject: [PATCH 04/17] Fixing some metadata --- .../pom.xml | 25 +------------------ 1 file changed, 1 insertion(+), 24 deletions(-) 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 index af240b9df..9d0e33498 100644 --- a/azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml +++ b/azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml @@ -5,7 +5,7 @@ spring-cloud-starter-stream-sink-wasb spring-cloud-starter-stream-sink-wasb - Spring Cloud Stream Windows Azure Storage Blob (WASB) Sink + Spring Cloud Stream Windows Azure Blob Storage Sink org.springframework.cloud.stream.app @@ -16,7 +16,6 @@ UTF-8 1.8 - 4.3.0 @@ -45,26 +44,4 @@ - - - - org.springframework.cloud - spring-cloud-dependencies - Brixton.SR1 - pom - import - - - - - - - - org.springframework.boot - spring-boot-maven-plugin - - - - - From f8b3644d8496c59c2d0d5d99f24e18ca9b287e0b Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 13:22:58 -0600 Subject: [PATCH 05/17] More pom cleanup --- .../pom.xml | 23 ++----------------- 1 file changed, 2 insertions(+), 21 deletions(-) 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 index 9d0e33498..7d2330552 100644 --- a/azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml +++ b/azure-storage/spring-cloud-starter-stream-sink-azure-blob/pom.xml @@ -15,33 +15,14 @@ UTF-8 - 1.8 - org.springframework.cloud - spring-cloud-starter-stream-kafka - - - org.springframework.boot - spring-boot-starter-test + org.springframework.cloud.stream.app + app-starters-test-support test - - org.apache.commons - commons-lang3 - 3.0 - - - org.slf4j - slf4j-api - - - org.springframework.boot - spring-boot-configuration-processor - true - From 18c5fa84a7fc7a5ae7d5e5ec202db4c45942b3b1 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 13:23:36 -0600 Subject: [PATCH 06/17] Fix javadoc switcheroo, shorten a few param names --- .../storage/sink/AzureBlobSinkProperties.java | 27 ++++++------------- 1 file changed, 8 insertions(+), 19 deletions(-) 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 index d0cf56427..6d1dbb5a3 100644 --- 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 @@ -30,20 +30,9 @@ * * @author Kyle Dunn */ -@ConfigurationProperties("azure-blob-sink") +@ConfigurationProperties("azure.blob") public class AzureBlobSinkProperties { - /** - * The name to use. - */ - @Value("${spring.application.name:azure-blob.sink}") - private String name; - - /** - * The level at which to log messages. - */ - private LoggingHandler.Level level = INFO; - /** * The default Azure endpoint protocol. */ @@ -52,30 +41,30 @@ public class AzureBlobSinkProperties { /** * The Azure Storage Account name. */ - private String accountName; + private String account; /** * The Azure Storage Account key. */ - private String accountKey; + private String key; /** * The Azure Storage Container name. */ - private String containerName; + private String container; /** * The Azure Storage Blob name. */ - private String blobName; + private String blob; /** - * Set container access policy to public. + * Create a container if it doesn't already exist. */ - private boolean autoCreateContainer = true; + private boolean createContainer = true; /** - * Create a container if it doesn't already exist. + * Set container access policy to public. */ private boolean publicPermission = true; From 24ca62f85c0a5ad651dc1f08ac5babf34646e0fc Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 13:24:09 -0600 Subject: [PATCH 07/17] Implementing some suggestions for generalizing and more correct error handling --- .../sink/AzureBlobSinkConfiguration.java | 23 +++++++------------ 1 file changed, 8 insertions(+), 15 deletions(-) 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 index 02483709a..16ba53019 100644 --- 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 @@ -85,7 +85,7 @@ public void setBlobService() { // Make the container public if (this.properties.getPublicPermission()) { - LOG.info("getBlobService() : making container publically accessible"); + LOG.info("getBlobService() : making container publicly accessible"); // Create a permissions object BlobContainerPermissions containerPermissions = new BlobContainerPermissions(); @@ -116,21 +116,14 @@ public void setBlobService() { } @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 void pushToAzureBlob(Message message) { + // Upload the payload to the blob + if (this.properties.getAppendOnly()) { + ((CloudAppendBlob) blobService).appendText(message.getPayload()); + } + else { + ((CloudBlockBlob) blobService).uploadText(message.getPayload()); } - } public static void main(String[] args) { From 6801bac5df023706e40a8c70247bfdd6a7e1f361 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 13:27:42 -0600 Subject: [PATCH 08/17] Must be a string since this is MVP - only using uploadText methods for now - TODO generalize for byte[] in next iteration --- .../app/azure/storage/sink/AzureBlobSinkConfiguration.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 index 16ba53019..d59b6a7b8 100644 --- 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 @@ -119,10 +119,10 @@ public void setBlobService() { public void pushToAzureBlob(Message message) { // Upload the payload to the blob if (this.properties.getAppendOnly()) { - ((CloudAppendBlob) blobService).appendText(message.getPayload()); + ((CloudAppendBlob) blobService).appendText(message.getPayload().toString()); } else { - ((CloudBlockBlob) blobService).uploadText(message.getPayload()); + ((CloudBlockBlob) blobService).uploadText(message.getPayload().toString()); } } From 3a15e21cc906d9773fdf2af29d7e8fc43bcb9284 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 13:28:05 -0600 Subject: [PATCH 09/17] Premature push - forgot to build first --- .../storage/sink/AzureBlobSinkProperties.java | 38 +++++-------------- 1 file changed, 10 insertions(+), 28 deletions(-) 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 index 6d1dbb5a3..c0adbe268 100644 --- 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 @@ -79,24 +79,6 @@ public class AzureBlobSinkProperties { */ private boolean overwriteExistingAppend = false; - @NotBlank - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - - @NotNull - public LoggingHandler.Level getLevel() { - return level; - } - - public void setLevel(LoggingHandler.Level level) { - this.level = level; - } - public String getDefaultEndpointsProtocol() { return defaultEndpointsProtocol; } @@ -106,44 +88,44 @@ public void setDefaultEndpointsProtocol(String p) { } public String getAccountName() { - return accountName; + return account; } public void setAccountName(String n) { - this.accountName = n; + this.account = n; } public String getAccountKey() { - return accountKey; + return key; } public void setAccountKey(String k) { - this.accountKey = k; + this.key = k; } public String getContainerName() { - return containerName; + return container; } public void setContainerName(String c) { - this.containerName = c; + this.container = c; } public String getBlobName() { - return blobName; + return blob; } public void setBlobName(String b) { - this.blobName = b; + this.blob = b; } public boolean getAutoCreateContainer() { - return autoCreateContainer; + return createContainer; } // Automatically create the container public void setAutoCreateContainer(Boolean b) { - this.autoCreateContainer = b; + this.createContainer = b; } public boolean getPublicPermission() { From 7993e4c5f5b10a1d01952aeeae8df7cccba514ec Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 13:33:04 -0600 Subject: [PATCH 10/17] A bit more cleanup, pure PR suggestions - commenting out LOG() calls for someone who knows what they're doing to reimplement --- .../sink/AzureBlobSinkConfiguration.java | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) 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 index d59b6a7b8..02efd6491 100644 --- 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 @@ -16,8 +16,7 @@ package org.springframework.cloud.stream.app.azure.storage.sink; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; +import java.io.IOException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; @@ -32,6 +31,7 @@ // 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; @@ -49,8 +49,6 @@ @SpringBootApplication public class AzureBlobSinkConfiguration { - protected static Logger LOG = LoggerFactory.getLogger(AzureBlobSinkConfiguration.class); - @Autowired private AzureBlobSinkProperties properties; @@ -68,7 +66,7 @@ public void setBlobService() { // Setup the cloud storage account. CloudStorageAccount account = CloudStorageAccount.parse(storageConnectionString); - LOG.info("getBlobService() : using account {}", this.properties.getAccountName()); + //LOG.info("getBlobService() : using account {}", this.properties.getAccountName()); // Create a blob service client CloudBlobClient blobClient = account.createCloudBlobClient(); @@ -77,7 +75,7 @@ public void setBlobService() { // The container name must be lower case CloudBlobContainer container = blobClient.getContainerReference(this.properties.getContainerName().toLowerCase()); - LOG.info("getBlobService() : using container {}", this.properties.getContainerName()); + //LOG.info("getBlobService() : using container {}", this.properties.getContainerName()); if (this.properties.getAutoCreateContainer()) { container.createIfNotExists(); @@ -85,7 +83,7 @@ public void setBlobService() { // Make the container public if (this.properties.getPublicPermission()) { - LOG.info("getBlobService() : making container publicly accessible"); + //LOG.info("getBlobService() : making container publicly accessible"); // Create a permissions object BlobContainerPermissions containerPermissions = new BlobContainerPermissions(); @@ -97,7 +95,7 @@ public void setBlobService() { container.uploadPermissions(containerPermissions); } - LOG.info("getBlobService() : using blob name {}", this.properties.getBlobName()); + //LOG.info("getBlobService() : using blob name {}", this.properties.getBlobName()); if (this.properties.getAppendOnly()) { this.blobService = container.getAppendBlobReference(this.properties.getBlobName()); @@ -110,13 +108,13 @@ public void setBlobService() { } } catch (Exception e) { // Log the stack trace. - LOG.error("getBlobService() : {}", e.getMessage()); + //LOG.error("getBlobService() : {}", e.getMessage()); //throw e; } } @ServiceActivator(inputChannel=Sink.INPUT) - public void pushToAzureBlob(Message message) { + public void pushToAzureBlob(Message message) throws StorageException, IOException { // Upload the payload to the blob if (this.properties.getAppendOnly()) { ((CloudAppendBlob) blobService).appendText(message.getPayload().toString()); From 8a2592d18b67b8bd2c4d54fc3f42660ec699ae2c Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 13:57:19 -0600 Subject: [PATCH 11/17] Adding new module starter to the root pom for SCS app starters --- pom.xml | 1 + 1 file changed, 1 insertion(+) 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 From af5b03e433283cedd2ed8e912920b32ae15de972 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 14:04:17 -0600 Subject: [PATCH 12/17] Adding Azure Storage sink reference to spring-cloud-stream-app-generator pom --- spring-cloud-stream-app-generator/pom.xml | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/spring-cloud-stream-app-generator/pom.xml b/spring-cloud-stream-app-generator/pom.xml index 9e2346d41..ef825f076 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.class + true + true From 628ccce08dcf2e230f28abcddbc9c1ded5db6d1e Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 14:10:11 -0600 Subject: [PATCH 13/17] Fixing incomplete reference --- spring-cloud-stream-app-generator/pom.xml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/spring-cloud-stream-app-generator/pom.xml b/spring-cloud-stream-app-generator/pom.xml index ef825f076..0221d19d9 100644 --- a/spring-cloud-stream-app-generator/pom.xml +++ b/spring-cloud-stream-app-generator/pom.xml @@ -44,7 +44,7 @@ org.springframework.cloud.stream.app.test.aggregate.counter.AggregateCounterSinkTestConfiguration.class - spring.cloud.starter.stream.sink.azure.blob.class + spring.cloud.starter.stream.sink.azure.blob.AzureBlobSinkConfiguration.class true From 0a93b174cd928e5f4269d60dbbb7bcdba157d847 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Mon, 22 Aug 2016 15:55:18 -0600 Subject: [PATCH 14/17] Fixinf exception catchall bad practice --- .../sink/AzureBlobSinkConfiguration.java | 74 +++++++++---------- 1 file changed, 35 insertions(+), 39 deletions(-) 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 index 02efd6491..a7ac18578 100644 --- 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 @@ -17,6 +17,8 @@ package org.springframework.cloud.stream.app.azure.storage.sink; import java.io.IOException; +import java.net.URISyntaxException; +import java.security.InvalidKeyException; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.SpringApplication; @@ -55,61 +57,55 @@ public class AzureBlobSinkConfiguration { private CloudBlob blobService; @Autowired - public void setBlobService() { + 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(); - try { - // Setup the cloud storage account. - CloudStorageAccount account = CloudStorageAccount.parse(storageConnectionString); + // Setup the cloud storage account. + CloudStorageAccount account = CloudStorageAccount.parse(storageConnectionString); - //LOG.info("getBlobService() : using account {}", this.properties.getAccountName()); + //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()); + // 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()); + //LOG.info("getBlobService() : using container {}", this.properties.getContainerName()); - if (this.properties.getAutoCreateContainer()) { - container.createIfNotExists(); - } + if (this.properties.getAutoCreateContainer()) { + container.createIfNotExists(); + } - // Make the container public - if (this.properties.getPublicPermission()) { - //LOG.info("getBlobService() : making container publicly accessible"); + // Make the container public + if (this.properties.getPublicPermission()) { + //LOG.info("getBlobService() : making container publicly accessible"); - // Create a permissions object - BlobContainerPermissions containerPermissions = new BlobContainerPermissions(); + // Create a permissions object + BlobContainerPermissions containerPermissions = new BlobContainerPermissions(); - // Include public access in the permissions object - containerPermissions.setPublicAccess(BlobContainerPublicAccessType.CONTAINER); + // Include public access in the permissions object + containerPermissions.setPublicAccess(BlobContainerPublicAccessType.CONTAINER); - // Set the permissions on the container - container.uploadPermissions(containerPermissions); - } + // 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()); + //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(); } - } catch (Exception e) { - // Log the stack trace. - //LOG.error("getBlobService() : {}", e.getMessage()); - //throw e; + } + else { + this.blobService = container.getBlockBlobReference(this.properties.getBlobName()); } } From 6dc6eb17ef8427f4c036562aeefb77d3d9bde8d4 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Wed, 12 Oct 2016 11:02:48 -0600 Subject: [PATCH 15/17] Commons logging --- .../azure/storage/sink/AzureBlobSinkConfiguration.java | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) 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 index a7ac18578..a99e5ed95 100644 --- 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 @@ -54,6 +54,8 @@ public class AzureBlobSinkConfiguration { @Autowired private AzureBlobSinkProperties properties; + private static Logger logger = LoggerFactory.getLogger(AzureBlobSinkConfiguration.class); + private CloudBlob blobService; @Autowired @@ -67,7 +69,7 @@ public void setBlobService() throws StorageException, URISyntaxException, Invali // Setup the cloud storage account. CloudStorageAccount account = CloudStorageAccount.parse(storageConnectionString); - //LOG.info("getBlobService() : using account {}", this.properties.getAccountName()); + logger.info("getBlobService() : using account {}", this.properties.getAccountName()); // Create a blob service client CloudBlobClient blobClient = account.createCloudBlobClient(); @@ -76,7 +78,7 @@ public void setBlobService() throws StorageException, URISyntaxException, Invali // The container name must be lower case CloudBlobContainer container = blobClient.getContainerReference(this.properties.getContainerName().toLowerCase()); - //LOG.info("getBlobService() : using container {}", this.properties.getContainerName()); + logger.info("getBlobService() : using container {}", this.properties.getContainerName()); if (this.properties.getAutoCreateContainer()) { container.createIfNotExists(); @@ -84,7 +86,7 @@ public void setBlobService() throws StorageException, URISyntaxException, Invali // Make the container public if (this.properties.getPublicPermission()) { - //LOG.info("getBlobService() : making container publicly accessible"); + logger.info("getBlobService() : making container publicly accessible"); // Create a permissions object BlobContainerPermissions containerPermissions = new BlobContainerPermissions(); @@ -96,7 +98,7 @@ public void setBlobService() throws StorageException, URISyntaxException, Invali container.uploadPermissions(containerPermissions); } - //LOG.info("getBlobService() : using blob name {}", this.properties.getBlobName()); + logger.info("getBlobService() : using blob name {}", this.properties.getBlobName()); if (this.properties.getAppendOnly()) { this.blobService = container.getAppendBlobReference(this.properties.getBlobName()); From dbf72d60f1535b5a37f2d0dc2ace85263794977d Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Wed, 12 Oct 2016 11:10:23 -0600 Subject: [PATCH 16/17] Commons logging import --- .../app/azure/storage/sink/AzureBlobSinkConfiguration.java | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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 index a99e5ed95..2cf2ebb4b 100644 --- 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 @@ -20,6 +20,9 @@ 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; @@ -54,7 +57,7 @@ public class AzureBlobSinkConfiguration { @Autowired private AzureBlobSinkProperties properties; - private static Logger logger = LoggerFactory.getLogger(AzureBlobSinkConfiguration.class); + private static Log logger = LogFactory.getLog(AzureBlobSinkConfiguration.class); private CloudBlob blobService; From 9ecff35df9770bbdd20b3683784d121611474648 Mon Sep 17 00:00:00 2001 From: Kyle Dunn Date: Wed, 12 Oct 2016 11:12:11 -0600 Subject: [PATCH 17/17] Commons logging syntax fixes --- .../app/azure/storage/sink/AzureBlobSinkConfiguration.java | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 index 2cf2ebb4b..05e92878b 100644 --- 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 @@ -72,7 +72,7 @@ public void setBlobService() throws StorageException, URISyntaxException, Invali // Setup the cloud storage account. CloudStorageAccount account = CloudStorageAccount.parse(storageConnectionString); - logger.info("getBlobService() : using account {}", this.properties.getAccountName()); + logger.info("getBlobService() : using account " + this.properties.getAccountName()); // Create a blob service client CloudBlobClient blobClient = account.createCloudBlobClient(); @@ -81,7 +81,7 @@ public void setBlobService() throws StorageException, URISyntaxException, Invali // The container name must be lower case CloudBlobContainer container = blobClient.getContainerReference(this.properties.getContainerName().toLowerCase()); - logger.info("getBlobService() : using container {}", this.properties.getContainerName()); + logger.info("getBlobService() : using container " + this.properties.getContainerName()); if (this.properties.getAutoCreateContainer()) { container.createIfNotExists(); @@ -101,7 +101,7 @@ public void setBlobService() throws StorageException, URISyntaxException, Invali container.uploadPermissions(containerPermissions); } - logger.info("getBlobService() : using blob name {}", this.properties.getBlobName()); + logger.info("getBlobService() : using blob name " + this.properties.getBlobName()); if (this.properties.getAppendOnly()) { this.blobService = container.getAppendBlobReference(this.properties.getBlobName());