Eg sync stack testproxy

eng/versioning/version_client.txt

@@ -441,6 +441,7 @@ com.azure.tools:azure-sdk-build-tool;1.0.0;1.1.0-beta.1
 # note: The unreleased dependencies will not be manipulated with the automatic PR creation code.
 # In the pom, the version update tag after the version should name the unreleased package and the dependency version:
 # <!-- {x-version-update;unreleased_com.azure:azure-core;dependency} -->
+unreleased_com.azure:azure-messaging-eventgrid;4.19.0-beta.1
 
 # Released Beta dependencies: Copy the entry from above, prepend "beta_", remove the current
 # version and set the version to the released beta. Released beta dependencies are only valid

sdk/core/azure-core-test/src/main/java/com/azure/core/test/TestProxyTestBase.java

@@ -5,15 +5,9 @@
 
 import com.azure.core.test.utils.TestProxyManager;
 import com.azure.core.util.logging.ClientLogger;
-import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.TestInfo;
 
-import java.nio.file.Path;
-import java.nio.file.Paths;
-
-import static com.azure.core.test.utils.TestUtils.toURI;
-
 /**
  * Base class for running live and playback tests using test-proxy
  */
@@ -30,8 +24,6 @@ public TestProxyTestBase() {
         super();
     }
 
-    private static TestProxyManager testProxyManager;
-
     /**
      * Before tests are executed, determines the test mode by reading the {@code AZURE_TEST_MODE} environment variable.
      * If it is not set, {@link TestMode#PLAYBACK}
@@ -40,20 +32,8 @@ public TestProxyTestBase() {
     @BeforeAll
     public static void setupTestProxy(TestInfo testInfo) {
         testMode = initializeTestMode();
-        Path testClassPath = Paths.get(toURI(testInfo.getTestClass().get().getResource(testInfo.getTestClass().get().getSimpleName() + ".class")));
         if (isTestProxyEnabled() && (testMode == TestMode.PLAYBACK || testMode == TestMode.RECORD)) {
-            testProxyManager = new TestProxyManager(testClassPath);
-            testProxyManager.startProxy();
-        }
-    }
-
-    /**
-     * Performs cleanup actions after all tests are executed.
-     */
-    @AfterAll
-    public static void teardownTestProxy() {
-        if (testProxyManager != null) {
-            testProxyManager.stopProxy();
+            TestProxyManager.startProxy();
         }
     }
 }

sdk/core/azure-core-test/src/main/java/com/azure/core/test/http/TestProxyPlaybackClient.java

@@ -14,6 +14,7 @@
 import com.azure.core.test.utils.HttpURLConnectionHttpClient;
 import com.azure.core.test.utils.TestProxyUtils;
 import com.azure.core.util.Context;
+import com.azure.core.util.logging.ClientLogger;
 import com.azure.core.util.serializer.JacksonAdapter;
 import com.azure.core.util.serializer.SerializerAdapter;
 import com.azure.core.util.serializer.SerializerEncoding;
@@ -32,6 +33,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
+import java.util.concurrent.TimeUnit;
 
 import static com.azure.core.test.implementation.TestingHelpers.X_RECORDING_FILE_LOCATION;
 import static com.azure.core.test.implementation.TestingHelpers.X_RECORDING_ID;
@@ -46,6 +48,7 @@
  */
 public class TestProxyPlaybackClient implements HttpClient {
 
+    private static final ClientLogger LOGGER = new ClientLogger(TestProxyPlaybackClient.class);
     private final HttpClient client;
     private final URL proxyUrl;
     private String xRecordingId;
@@ -89,11 +92,7 @@ public Queue<String> startPlayback(File recordFile, Path testClassPath) {
                     SerializerEncoding.JSON))
                 .setHeader(HttpHeaderName.ACCEPT, "application/json")
                 .setHeader(HttpHeaderName.CONTENT_TYPE, "application/json");
-        } catch (IOException e) {
-            throw new RuntimeException(e);
-        }
-
-        try (HttpResponse response = client.sendSync(request, Context.NONE)) {
+            HttpResponse response = sendRequestWithRetries(request);
             checkForTestProxyErrors(response);
             xRecordingId = response.getHeaderValue(X_RECORDING_ID);
             xRecordingFileLocation = new String(Base64.getUrlDecoder().decode(
@@ -127,13 +126,42 @@ public Queue<String> startPlayback(File recordFile, Path testClassPath) {
         }
     }
 
+    private HttpResponse sendRequestWithRetries(HttpRequest request) {
+        int retries = 0;
+        while (true) {
+            try {
+                HttpResponse response = client.sendSync(request, Context.NONE);
+                if (response.getStatusCode() / 100 != 2) {
+                    throw new RuntimeException("Test proxy returned a non-successful status code. "
+                        + response.getStatusCode() + "; response: " + response.getBodyAsString().block());
+                }
+                return response;
+            } catch (Exception e) {
+                retries++;
+                if (retries >= 3) {
+                    throw e;
+                }
+                sleep(1);
+                LOGGER.warning("Retrying request to test proxy. Retry attempt: " + retries);
+            }
+        }
+    }
+
+    private void sleep(int durationInSeconds) {
+        try {
+            TimeUnit.SECONDS.sleep(durationInSeconds);
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
     /**
      * Stops playback of a test recording.
      */
     public void stopPlayback() {
         HttpRequest request = new HttpRequest(HttpMethod.POST, proxyUrl + "/playback/stop")
             .setHeader(X_RECORDING_ID, xRecordingId);
-        client.sendSync(request, Context.NONE).close();
+        sendRequestWithRetries(request);
     }
 
     /**
@@ -210,7 +238,7 @@ public void addMatcherRequests(List<TestProxyRequestMatcher> matchers) {
             }
             matcherRequests.forEach(request -> {
                 request.setHeader(X_RECORDING_ID, xRecordingId);
-                client.sendSync(request, Context.NONE).close();
+                sendRequestWithRetries(request);
             });
         } else {
             this.matchers.addAll(matchers);

sdk/core/azure-core-test/src/main/java/com/azure/core/test/utils/TestProxyManager.java

@@ -24,37 +24,34 @@
 /**
  * Manages running the test recording proxy server
  */
-public class TestProxyManager {
+public final class TestProxyManager {
     private static final ClientLogger LOGGER = new ClientLogger(TestProxyManager.class);
-    private Process proxy;
-    private final Path testClassPath;
+    private static Process proxy;
+    private static final Path WORKING_DIRECTORY = Paths.get(System.getProperty("user.dir"));
 
-    /**
-     * Construct a {@link TestProxyManager} for controlling the external test proxy.
-     * @param testClassPath the test class path
-     */
-    public TestProxyManager(Path testClassPath) {
-        this.testClassPath = testClassPath;
-        // This is necessary to stop the proxy when the debugger is stopped.
-        Runtime.getRuntime().addShutdownHook(new Thread(this::stopProxy));
+    static {
+        Runtime.getRuntime().addShutdownHook(new Thread(TestProxyManager::stopProxy));
         if (runningLocally()) {
-            TestProxyDownloader.installTestProxy(testClassPath);
+            TestProxyDownloader.installTestProxy(WORKING_DIRECTORY);
         }
     }
 
+    @Deprecated
+    private TestProxyManager() { }
+
     /**
      * Start an instance of the test proxy.
      * @throws UncheckedIOException There was an issue communicating with the proxy.
      * @throws RuntimeException There was an issue starting the proxy process.
      */
-    public void startProxy() {
+    public static synchronized void startProxy() {
         try {
             // if we're not running in CI we will check to see if someone has started the proxy, and start one if not.
             if (runningLocally() && !checkAlive(1, Duration.ofSeconds(1), null)) {
                 String commandLine = Paths.get(TestProxyDownloader.getProxyDirectory().toString(),
                     TestProxyUtils.getProxyProcessName()).toString();
 
-                Path repoRoot = TestUtils.getRepoRootResolveUntil(testClassPath, "eng");
+                Path repoRoot = TestUtils.getRepoRootResolveUntil(WORKING_DIRECTORY, "eng");
 
                 // Resolve the path to the repo root 'target' folder and create the folder if it doesn't exist.
                 // This folder will be used to store the 'test-proxy.log' file to enable simpler debugging of Test Proxy
@@ -128,7 +125,7 @@ private static boolean checkAlive(int loops, Duration waitTime, Process proxy) t
     /**
      * Stop the running instance of the test proxy.
      */
-    public void stopProxy() {
+    private static void stopProxy() {
         if (proxy != null && proxy.isAlive()) {
             proxy.destroy();
         }
@@ -138,7 +135,7 @@ public void stopProxy() {
      * Checks the environment variables commonly set in CI to determine if the run is local.
      * @return True if the run is local.
      */
-    private boolean runningLocally() {
+    private static boolean runningLocally() {
         return Configuration.getGlobalConfiguration().get("TF_BUILD") == null
             && Configuration.getGlobalConfiguration().get("CI") == null;
     }

sdk/eventgrid/azure-messaging-eventgrid/CHANGELOG.md

@@ -18,7 +18,7 @@
 ### Other Changes
 
 #### Dependency Updates
-- 
+
 - Upgraded `azure-core` from `1.42.0` to version `1.43.0`.
 - Upgraded `azure-core-http-netty` from `1.13.6` to version `1.13.7`.
 

sdk/eventgrid/azure-messaging-eventgrid/assets.json

@@ -0,0 +1,6 @@
+{
+  "AssetsRepo": "Azure/azure-sdk-assets",
+  "AssetsRepoPrefixPath": "java",
+  "TagPrefix": "java/eventgrid/azure-messaging-eventgrid",
+  "Tag": "java/eventgrid/azure-messaging-eventgrid_7bb36a1579"
+}

sdk/eventgrid/azure-messaging-eventgrid/pom.xml

@@ -56,6 +56,8 @@
       --add-opens com.azure.messaging.eventgrid/com.azure.messaging.eventgrid.implementation=ALL-UNNAMED
 
       --add-reads com.azure.messaging.eventgrid=com.azure.core.serializer.json.jackson
+      --add-exports com.azure.core/com.azure.core.implementation.util=ALL-UNNAMED
+      --add-exports com.azure.core/com.azure.core.implementation=ALL-UNNAMED
     </javaModulesSurefireArgLine>
   </properties>
 

sdk/eventgrid/azure-messaging-eventgrid/src/main/java/com/azure/messaging/eventgrid/EventGridPublisherAsyncClient.java

@@ -8,14 +8,11 @@
 import com.azure.core.annotation.ServiceMethod;
 import com.azure.core.credential.AzureKeyCredential;
 import com.azure.core.credential.AzureSasCredential;
-import com.azure.core.http.HttpHeaders;
 import com.azure.core.http.HttpPipeline;
-import com.azure.core.http.policy.AddHeadersFromContextPolicy;
 import com.azure.core.http.rest.Response;
 import com.azure.core.models.CloudEvent;
 import com.azure.core.util.BinaryData;
 import com.azure.core.util.Context;
-import com.azure.core.util.CoreUtils;
 import com.azure.core.util.logging.ClientLogger;
 import com.azure.core.util.tracing.Tracer;
 import com.azure.messaging.eventgrid.implementation.Constants;
@@ -39,7 +36,6 @@
 import java.util.Base64;
 import java.util.Collections;
 import java.util.List;
-import java.util.Map;
 import java.util.Objects;
 
 import static com.azure.core.util.FluxUtil.monoError;
@@ -155,7 +151,6 @@
 @ServiceClient(builder = EventGridPublisherClientBuilder.class, isAsync = true)
 public final class EventGridPublisherAsyncClient<T> {
 
-    private static final String PARTNER_CHANNEL_HEADER_NAME = "aeg-channel-name";
     private final String hostname;
 
     private final EventGridPublisherClientImpl impl;
@@ -312,25 +307,9 @@ Mono<Response<Void>> sendEventsWithResponse(Iterable<T> events, String channelNa
         if (context == null) {
             context = Context.NONE;
         }
-        if (!CoreUtils.isNullOrEmpty(channelName)) {
-            String requestHttpHeadersKey = AddHeadersFromContextPolicy.AZURE_REQUEST_HTTP_HEADERS_KEY;
-            Map<Object, Object> keyValues = context.getValues();
-            if (keyValues != null && keyValues.containsKey(requestHttpHeadersKey)) {
-                // if the given Context instance already contains custom headers,
-                // add partner channel header to HttpHeaders
-                Object value = keyValues.get(requestHttpHeadersKey);
-                if (value instanceof HttpHeaders) {
-                    HttpHeaders headers = (HttpHeaders) value;
-                    headers.add(PARTNER_CHANNEL_HEADER_NAME, channelName);
-                }
-            } else {
-                context = context.addData(requestHttpHeadersKey,
-                        new HttpHeaders().add(PARTNER_CHANNEL_HEADER_NAME, channelName));
-            }
-        }
 
         if (this.eventClass == CloudEvent.class) {
-            return this.sendCloudEventsWithResponse((Iterable<CloudEvent>) events, context);
+            return this.sendCloudEventsWithResponse((Iterable<CloudEvent>) events, channelName, context);
         } else if (this.eventClass == EventGridEvent.class) {
             return this.sendEventGridEventsWithResponse((Iterable<EventGridEvent>) events, context);
         } else {
@@ -395,15 +374,15 @@ Mono<Response<Void>> sendEventGridEventsWithResponse(Iterable<EventGridEvent> ev
             .flatMap(list -> this.impl.publishEventGridEventsWithResponseAsync(this.hostname, list, finalContext));
     }
 
-    Mono<Response<Void>> sendCloudEventsWithResponse(Iterable<CloudEvent> events, Context context) {
+    Mono<Response<Void>> sendCloudEventsWithResponse(Iterable<CloudEvent> events, String channelName, Context context) {
         if (events == null) {
             return monoError(logger, new NullPointerException("'events' cannot be null."));
         }
         final Context finalContext = context != null ? context : Context.NONE;
         this.addCloudEventTracePlaceHolder(events);
         return Flux.fromIterable(events)
             .collectList()
-            .flatMap(list -> this.impl.publishCloudEventEventsWithResponseAsync(this.hostname, list, null, finalContext));
+            .flatMap(list -> this.impl.publishCloudEventEventsWithResponseAsync(this.hostname, list, channelName, finalContext));
     }
 
     Mono<Response<Void>> sendCustomEventsWithResponse(Iterable<BinaryData> events, Context context) {