Merge branch 'storage-post-review1-dev' into storage-file

.github/CODEOWNERS

@@ -0,0 +1 @@
+# See for instructions on this file https://help.github.com/articles/about-codeowners/

README.md

@@ -29,11 +29,11 @@ New wave of packages that we are currently releasing in **preview**. These libra
 These preview libraries can be easily identified by their folder, package, and namespaces names starting with `azure-`, e.g. `azure-keyvault`.
 
 The libraries released in the July 2019 preview:
-- [App Configuration](appconfiguration/client/README.md)
-- [Event Hubs](eventhubs/client/README.md)
-- [Identity](sdk/identity/azure-identity)
-- [Key Vault Keys](keyvault/client/keys/README.md)
-- [Key Vault Secrets](keyvault/client/secrets/README.md)
+- [App Configuration](sdk/appconfiguration/azure-data-appconfiguration/README.md)
+- [Event Hubs](sdk/eventhubs/azure-eventhubs/README.md)
+- [Identity](sdk/identity/azure-identity/README.md)
+- [Key Vault Keys](sdk/keyvault/azure-keyvault-keys/README.md)
+- [Key Vault Secrets](sdk/keyvault/client/azure-keyvault-secrets/README.md)
 - [Storage Blobs](storage/client/README.md)
 
 >NOTE: If you need to ensure your code is ready for production, use one of the stable libraries.

core/azure-core-test/src/main/java/com/azure/core/test/http/package-info.java

@@ -2,6 +2,6 @@
 // Licensed under the MIT License.
 
 /**
- * Package contains classes to test HTTP communications in Azure client libraries.
+ * Package containing classes to test HTTP communications in Azure client libraries.
  */
 package com.azure.core.test.http;

core/azure-core-test/src/main/java/com/azure/core/test/models/package-info.java

@@ -1,6 +1,6 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 /**
- * Package contains models used to test Azure client libraries.
+ * Package containing models used to test Azure client libraries.
  */
 package com.azure.core.test.models;

core/azure-core-test/src/main/java/com/azure/core/test/package-info.java

@@ -1,6 +1,6 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 /**
- * Package contains common test classes for Azure client libraries.
+ * Package containing common test classes for Azure client libraries.
  */
 package com.azure.core.test;

core/azure-core-test/src/main/java/com/azure/core/test/policy/package-info.java

@@ -1,7 +1,7 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 /**
- * Package contains {@link com.azure.core.http.policy.HttpPipelinePolicy HttpPipelinePolicies} used to test Azure
+ * Package containing {@link com.azure.core.http.policy.HttpPipelinePolicy HttpPipelinePolicies} used to test Azure
  * client libraries.
  */
 package com.azure.core.test.policy;

core/azure-core-test/src/main/java/com/azure/core/test/utils/package-info.java

@@ -1,6 +1,6 @@
 // Copyright (c) Microsoft Corporation. All rights reserved.
 // Licensed under the MIT License.
 /**
- * Package contains utility classes used for testing Azure client libraries.
+ * Package containing utility classes used for testing Azure client libraries.
  */
 package com.azure.core.test.utils;

core/azure-core/src/main/java/com/azure/core/implementation/util/FluxUtil.java

@@ -3,11 +3,22 @@
 
 package com.azure.core.implementation.util;
 
+import com.azure.core.util.Context;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.CompositeByteBuf;
 import io.netty.buffer.Unpooled;
 import io.netty.util.ReferenceCountUtil;
+import java.io.IOException;
+import java.lang.reflect.Type;
+import java.nio.channels.AsynchronousFileChannel;
+import java.nio.channels.CompletionHandler;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 import org.reactivestreams.Subscriber;
 import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
@@ -16,13 +27,6 @@
 import reactor.core.publisher.MonoSink;
 import reactor.core.publisher.Operators;
 
-import java.io.IOException;
-import java.lang.reflect.Type;
-import java.nio.channels.AsynchronousFileChannel;
-import java.nio.channels.CompletionHandler;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
-import java.util.concurrent.atomic.AtomicLongFieldUpdater;
-
 /**
  * Utility type exposing methods to deal with {@link Flux}.
  */
@@ -174,6 +178,64 @@ public static Mono<Void> bytebufStreamToFile(Flux<ByteBuf> content, Asynchronous
         return Mono.create(emitter -> content.subscribe(new ByteBufToFileSubscriber(outFile, position, emitter)));
     }
 
+    /**
+     * This method converts the incoming {@code subscriberContext} from {@link reactor.util.context.Context Reactor
+     * Context} to {@link Context Azure Context} and calls the given lambda function with this context and returns a
+     * single entity of type {@code T}
+     * <p>
+     *  If the reactor context is empty, {@link Context#NONE} will be used to call the lambda function
+     * </p>
+     *
+     * <p><strong>Code samples</strong></p>
+     * {@codesnippet com.azure.core.implementation.util.fluxutil.monocontext}
+     *
+     * @param serviceCall The lambda function that makes the service call into which azure context will be passed
+     * @param <T> The type of response returned from the service call
+     * @return The response from service call
+     */
+    public static <T> Mono<T> monoContext(Function<Context, Mono<T>> serviceCall) {
+        return Mono.subscriberContext()
+            .map(FluxUtil::toAzureContext)
+            .flatMap(serviceCall);
+    }
+
+    /**
+     * This method converts the incoming {@code subscriberContext} from {@link reactor.util.context.Context Reactor
+     * Context} to {@link Context Azure Context} and calls the given lambda function with this context and returns a
+     * collection of type {@code T}
+     * <p>
+     *  If the reactor context is empty, {@link Context#NONE} will be used to call the lambda function
+     * </p>
+     *
+     *  <p><strong>Code samples</strong></p>
+     *  {@codesnippet com.azure.core.implementation.util.fluxutil.fluxcontext}
+     *
+     * @param serviceCall The lambda function that makes the service call into which the context will be passed
+     * @param <T> The type of response returned from the service call
+     * @return The response from service call
+     */
+    public static <T> Flux<T> fluxContext(Function<Context, Flux<T>> serviceCall) {
+        return Mono.subscriberContext()
+            .map(FluxUtil::toAzureContext)
+            .flatMapMany(serviceCall);
+    }
+
+    /**
+     * Converts a reactor context to azure context. If the reactor context is {@code null} or empty,
+     * {@link Context#NONE} will be returned.
+     *
+     * @param context The reactor context
+     * @return The azure context
+     */
+    private static Context toAzureContext(reactor.util.context.Context context) {
+        Map<Object, Object> keyValues = context.stream()
+            .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
+        if (ImplUtils.isNullOrEmpty(keyValues)) {
+            return Context.NONE;
+        }
+        return Context.of(keyValues);
+    }
+
     private static class ByteBufToFileSubscriber implements Subscriber<ByteBuf> {
         private ByteBufToFileSubscriber(AsynchronousFileChannel outFile, long position, MonoSink<Void> emitter) {
             this.outFile = outFile;

core/azure-core/src/main/java/com/azure/core/util/Context.java

@@ -4,6 +4,8 @@
 package com.azure.core.util;
 
 import com.azure.core.implementation.annotation.Immutable;
+import com.azure.core.implementation.util.ImplUtils;
+import java.util.Map;
 import java.util.Optional;
 
 /**
@@ -63,6 +65,30 @@ public Context addData(Object key, Object value) {
         return new Context(this, key, value);
     }
 
+    /**
+     * Creates a new immutable {@link Context} object with all the keys and values provided by
+     * the input {@link Map}
+     *
+     * @param keyValues The input key value pairs that will be added to this context
+     * @return Context object containing all the key-value pairs in the input map
+     * @throws IllegalArgumentException If {@code keyValues} is {@code null} or empty
+     */
+    public static Context of(Map<Object, Object> keyValues) {
+        if (ImplUtils.isNullOrEmpty(keyValues)) {
+            throw new IllegalArgumentException("Key value map cannot be null or empty");
+        }
+
+        Context context = null;
+        for (Map.Entry<Object, Object> entry : keyValues.entrySet()) {
+            if (context == null) {
+                context = new Context(entry.getKey(), entry.getValue());
+            } else {
+                context = context.addData(entry.getKey(), entry.getValue());
+            }
+        }
+        return context;
+    }
+
     /**
      * Scans the linked-list of {@link Context} objects looking for one with the specified key.
      * Note that the first key found, i.e. the most recently added, will be returned.

core/azure-core/src/samples/java/com/azure/core/implementation/util/FluxUtilJavaDocCodeSnippets.java

@@ -0,0 +1,57 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.implementation.util;
+
+import com.azure.core.util.Context;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+/**
+ * Code snippets for {@link FluxUtil}
+ */
+public class FluxUtilJavaDocCodeSnippets {
+
+    /**
+     * Code snippet for using {@link FluxUtil} with single item response
+     */
+    public void codeSnippetForCallWithSingleResponse() {
+        // BEGIN: com.azure.core.implementation.util.fluxutil.monocontext
+        String prefix = "Hello, ";
+        Mono<String> response = FluxUtil
+            .monoContext(context -> serviceCallReturnsSingle(prefix, context));
+        // END: com.azure.core.implementation.util.fluxutil.monocontext
+    }
+
+    /**
+     * Code snippet for using {@link FluxUtil} with collection response
+     */
+    public void codeSnippetForCallWithCollectionResponse() {
+        // BEGIN: com.azure.core.implementation.util.fluxutil.fluxcontext
+        String prefix = "Hello, ";
+        Flux<String> response = FluxUtil
+            .fluxContext(context -> serviceCallReturnsCollection(prefix, context));
+        // END: com.azure.core.implementation.util.fluxutil.fluxcontext
+    }
+
+    /**
+     * Implementation not provided
+     * @param prefix The prefix
+     * @param context Azure context
+     * @return {@link Flux#empty() empty} response
+     */
+    private Flux<String> serviceCallReturnsCollection(String prefix, Context context) {
+        return Flux.empty();
+    }
+
+    /**
+     * Implementation not provided
+     * @param prefix The prefix
+     * @param context Azure context
+     * @return {@link Mono#empty() empty} response
+     */
+    private Mono<String> serviceCallReturnsSingle(String prefix, Context context) {
+        return Mono.empty();
+    }
+
+}

core/azure-core/src/test/java/com/azure/core/implementation/util/FluxUtilTests.java

@@ -3,17 +3,15 @@
 
 package com.azure.core.implementation.util;
 
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import com.azure.core.util.Context;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.ByteBufUtil;
 import io.netty.buffer.Unpooled;
 import io.netty.util.ReferenceCountUtil;
-import org.junit.Ignore;
-import org.junit.Test;
-import reactor.core.Exceptions;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-import reactor.test.StepVerifier;
-
 import java.io.BufferedOutputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.File;
@@ -24,10 +22,16 @@
 import java.nio.file.StandardOpenOption;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
-
-import static org.junit.Assert.assertArrayEquals;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+import reactor.core.Exceptions;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
 
 public class FluxUtilTests {
 
@@ -262,6 +266,51 @@ public void testCollectByteBufStream() {
             0, 0, 0, (byte) 255,
             0, 0, 1, 0}, bytes);
     }
+
+    @Test
+    public void testCallWithContextGetSingle() {
+        String response = getSingle("Hello, ")
+            .subscriberContext(reactor.util.context.Context.of("FirstName", "Foo", "LastName", "Bar"))
+            .block();
+        Assert.assertEquals("Hello, Foo Bar", response);
+    }
+
+    @Test
+    public void testCallWithContextGetCollection() {
+        List<String> expectedLines = Arrays.asList("Hello,", "Foo", "Bar");
+        List<String> actualLines = new ArrayList<>();
+        getCollection("Hello, ")
+            .subscriberContext(reactor.util.context.Context.of("FirstName", "Foo", "LastName", "Bar"))
+            .doOnNext(line -> actualLines.add(line))
+            .subscribe();
+        Assert.assertEquals(expectedLines, actualLines);
+    }
+
+    private Mono<String> getSingle(String prefix) {
+        return FluxUtil.monoContext(context -> serviceCallSingle(prefix, context));
+    }
+
+    private Flux<String> getCollection(String prefix) {
+        return FluxUtil
+            .fluxContext(context -> serviceCallCollection(prefix, context));
+    }
+
+    private Mono<String> serviceCallSingle(String prefix, Context context) {
+        String msg = prefix
+            + context.getData("FirstName").orElse("Stranger")
+            + " "
+            + context.getData("LastName").orElse("");
+        return Mono.just(msg);
+    }
+
+    private Flux<String> serviceCallCollection(String prefix, Context context) {
+        String msg = prefix
+            + context.getData("FirstName").orElse("Stranger")
+            + " "
+            + context.getData("LastName").orElse("");
+
+        return Flux.just(msg.split(" "));
+    }
 //
     private static byte[] toBytes(ByteBuf bb) {
         byte[] bytes = new byte[bb.readableBytes()];