OpAMP WebSocket service

dependencyManagement/build.gradle.kts

@@ -57,5 +57,6 @@ dependencies {
     api("tools.profiler:async-profiler:4.0")
     api("com.blogspot.mydailyjava:weak-lock-free:0.18")
     api("org.agrona:agrona:1.22.0")
+    api("com.google.protobuf:protobuf-java-util:4.29.1")
   }
 }

opamp-client/build.gradle.kts

@@ -16,6 +16,7 @@ dependencies {
   annotationProcessor("com.google.auto.value:auto-value")
   compileOnly("com.google.auto.value:auto-value-annotations")
   testImplementation("org.mockito:mockito-inline")
+  testImplementation("com.google.protobuf:protobuf-java-util")
 }
 
 val opampReleaseInfo = tasks.register<Download>("opampLastReleaseInfo") {

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/connectivity/websocket/OkHttpWebSocket.java

@@ -0,0 +1,116 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.connectivity.websocket;
+
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import okhttp3.OkHttpClient;
+import okhttp3.Response;
+import okhttp3.WebSocketListener;
+import okio.ByteString;
+
+public class OkHttpWebSocket implements WebSocket {
+  private final String url;
+  private final OkHttpClient client;
+  private final AtomicBoolean starting = new AtomicBoolean(false);
+  private final AtomicBoolean closing = new AtomicBoolean(false);
+  private final AtomicBoolean running = new AtomicBoolean(false);
+  private final AtomicReference<okhttp3.WebSocket> webSocket = new AtomicReference<>();
+
+  public static OkHttpWebSocket create(String url) {
+    return create(url, new OkHttpClient());
+  }
+
+  public static OkHttpWebSocket create(String url, OkHttpClient client) {
+    return new OkHttpWebSocket(url, client);
+  }
+
+  private OkHttpWebSocket(String url, OkHttpClient client) {
+    this.url = url;
+    this.client = client;
+  }
+
+  @Override
+  public void open(Listener listener) {
+    if (running.get()) {
+      return;
+    }
+    if (starting.compareAndSet(false, true)) {
+      okhttp3.Request request = new okhttp3.Request.Builder().url(url).build();
+      webSocket.set(client.newWebSocket(request, new ListenerAdapter(listener)));
+    }
+  }
+
+  @Override
+  public boolean send(byte[] request) {
+    if (!running.get()) {
+      return false;
+    }
+    return getWebSocket().send(ByteString.of(request));
+  }
+
+  @Override
+  public void close(int code, @Nullable String reason) {
+    if (!running.get()) {
+      return;
+    }
+    if (closing.compareAndSet(false, true)) {
+      if (!getWebSocket().close(code, reason)) {
+        closing.set(false);
+        running.set(false);
+      }
+    }
+  }
+
+  private okhttp3.WebSocket getWebSocket() {
+    return Objects.requireNonNull(webSocket.get());
+  }
+
+  private class ListenerAdapter extends WebSocketListener {
+    private final Listener delegate;
+
+    private ListenerAdapter(Listener delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public void onOpen(@Nonnull okhttp3.WebSocket webSocket, @Nonnull Response response) {
+      running.set(true);
+      starting.set(false);
+      delegate.onOpen();
+    }
+
+    @Override
+    public void onClosing(@Nonnull okhttp3.WebSocket webSocket, int code, @Nonnull String reason) {
+      running.set(false);
+      closing.set(true);
+    }
+
+    @Override
+    public void onClosed(@Nonnull okhttp3.WebSocket webSocket, int code, @Nonnull String reason) {
+      running.set(false);
+      closing.set(false);
+      delegate.onClosed();
+    }
+
+    @Override
+    public void onFailure(
+        @Nonnull okhttp3.WebSocket webSocket, @Nonnull Throwable t, @Nullable Response response) {
+      running.set(false);
+      starting.set(false);
+      closing.set(false);
+      delegate.onFailure(t);
+    }
+
+    @Override
+    public void onMessage(@Nonnull okhttp3.WebSocket webSocket, @Nonnull ByteString bytes) {
+      delegate.onMessage(bytes.toByteArray());
+    }
+  }
+}

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/connectivity/websocket/WebSocket.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.opamp.client.internal.connectivity.websocket;
+
+import javax.annotation.Nullable;
+
+public interface WebSocket {
+  /**
+   * Starts the websocket connection if it's not yet started or if it has been closed.
+   *
+   * @param listener Will receive events from the websocket connection.
+   */
+  void open(Listener listener);
+
+  /**
+   * Stops the websocket connection if running. Nothing will happen if it's already stopped.
+   *
+   * @param code Status code as defined by <a
+   *     href="http://tools.ietf.org/html/rfc6455#section-7.4">Section 7.4 of RFC 6455</a>
+   * @param reason Reason for shutting down, as explained in <a
+   *     href="https://datatracker.ietf.org/doc/html/rfc6455#section-5.5.1">Section 5.5.1 of RFC
+   *     6455</a>
+   */
+  void close(int code, @Nullable String reason);
+
+  /**
+   * Sends a message via the websocket connection.
+   *
+   * @param request The message payload.
+   * @return {@link Boolean#FALSE} If the message can't get dispatched for any reason, whether the
+   *     websocket isn't running, or the connection isn't established, or it's terminated. {@link
+   *     Boolean#TRUE} if the message can get sent. Returning {@link Boolean#TRUE} doesn't guarantee
+   *     that the message will arrive at the remote peer.
+   */
+  boolean send(byte[] request);
+
+  interface Listener {
+
+    /**
+     * Called when the websocket connection is successfully established with the remote peer and may
+     * start sending messages.
+     */
+    void onOpen();
+
+    /** Called when the connection is terminated and no further messages can get transmitted. */
+    void onClosed();
+
+    /**
+     * Called when receiving a message from the remote peer.
+     *
+     * @param data The payload sent by the remote peer.
+     */
+    void onMessage(byte[] data);
+
+    /**
+     * Called when the connection is closed or cannot be established due to an error. No messages
+     * can get transmitted after this method is called.
+     *
+     * @param t The error.
+     */
+    void onFailure(Throwable t);
+  }
+}