[opamp-client] Refactor opamp client callbacks api

opamp-client/README.md

@@ -11,16 +11,14 @@ client [spec](https://github.com/open-telemetry/opamp-spec/blob/main/specificati
 
 ```java
 // Initializing it
-
 RequestService requestService = HttpRequestService.create(OkHttpSender.create("[OPAMP_SERVICE_URL]"));
 // RequestService requestService = WebSocketRequestService.create(OkHttpWebSocket.create("[OPAMP_SERVICE_URL]")); // Use this instead to connect to the server via WebSocket.
 OpampClient client =
     OpampClient.builder()
         .putIdentifyingAttribute("service.name", "My service name")
         .enableRemoteConfig()
         .setRequestService(requestService)
-        .build(
-            new OpampClient.Callbacks() {
+        .build(cx -> new OpampClient.Callbacks() {
               @Override
               public void onConnect() {}
 
@@ -37,7 +35,7 @@ OpampClient client =
                   // A remote config was received
 
                   // After applying it...
-                  client.setRemoteConfigStatus(
+                  cx.setRemoteConfigStatus(
                       new RemoteConfigStatus.Builder()
                           .status(RemoteConfigStatuses.RemoteConfigStatuses_APPLIED)
                           .build());

opamp-client/src/main/java/io/opentelemetry/opamp/client/OpampClient.java

@@ -35,6 +35,21 @@ static OpampClientBuilder builder() {
    */
   void setRemoteConfigStatus(RemoteConfigStatus remoteConfigStatus);
 
+  Callbacks NOOP_CALLBACKS =
+      new Callbacks() {
+        @Override
+        public void onConnect() {}
+
+        @Override
+        public void onConnectFailed(@org.jetbrains.annotations.Nullable Throwable throwable) {}
+
+        @Override
+        public void onErrorResponse(ServerErrorResponse errorResponse) {}
+
+        @Override
+        public void onMessage(MessageData messageData) {}
+      };
+
   interface Callbacks {
     /**
      * Called when the connection is successfully established to the Server. For WebSocket clients

opamp-client/src/main/java/io/opentelemetry/opamp/client/OpampClientBuilder.java

@@ -20,6 +20,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
+import java.util.function.Function;
 import javax.annotation.Nullable;
 import opamp.proto.AgentCapabilities;
 import opamp.proto.AgentDescription;
@@ -376,7 +377,7 @@ public OpampClientBuilder setEffectiveConfigState(State.EffectiveConfig effectiv
     return this;
   }
 
-  public OpampClient build(OpampClient.Callbacks callbacks) {
+  public OpampClient build(Function<OpampClient, OpampClient.Callbacks> callbacks) {
     List<KeyValue> protoIdentifyingAttributes = new ArrayList<>();
     List<KeyValue> protoNonIdentifyingAttributes = new ArrayList<>();
     identifyingAttributes.forEach(

opamp-client/src/main/java/io/opentelemetry/opamp/client/internal/impl/OpampClientImpl.java

@@ -29,6 +29,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Function;
 import java.util.function.Supplier;
 import javax.annotation.Nonnull;
 import okio.ByteString;
@@ -50,7 +51,7 @@ public final class OpampClientImpl
   private final OpampClientState state;
   private final RecipeManager recipeManager;
   private final AtomicBoolean hasStopped = new AtomicBoolean(false);
-  private final Callbacks callbacks;
+  private volatile Callbacks callbacks = NOOP_CALLBACKS;
 
   /** Fields that must always be sent. */
   private static final List<Field> REQUIRED_FIELDS;
@@ -82,7 +83,9 @@ public final class OpampClientImpl
   }
 
   public static OpampClientImpl create(
-      RequestService requestService, OpampClientState state, Callbacks callbacks) {
+      RequestService requestService,
+      OpampClientState state,
+      Function<OpampClient, Callbacks> callbacksFunction) {
     AgentToServerAppenders appenders =
         new AgentToServerAppenders(
             AgentDescriptionAppender.create(state.agentDescription),
@@ -95,7 +98,8 @@ public static OpampClientImpl create(
             AgentDisconnectAppender.create());
     OpampClientImpl client =
         new OpampClientImpl(
-            requestService, appenders, state, RecipeManager.create(REQUIRED_FIELDS), callbacks);
+            requestService, appenders, state, RecipeManager.create(REQUIRED_FIELDS));
+    client.callbacks = callbacksFunction.apply(client);
 
     // Start
     requestService.start(client, client);
@@ -110,13 +114,11 @@ private OpampClientImpl(
       RequestService requestService,
       AgentToServerAppenders appenders,
       OpampClientState state,
-      RecipeManager recipeManager,
-      Callbacks callbacks) {
+      RecipeManager recipeManager) {
     this.requestService = requestService;
     this.appenders = appenders;
     this.state = state;
     this.recipeManager = recipeManager;
-    this.callbacks = callbacks;
   }
 
   @Override

opamp-client/src/test/java/io/opentelemetry/opamp/client/internal/impl/OpampClientImplTest.java

@@ -395,7 +395,7 @@ private RecordedRequest initializeClient(ServerToAgent initialResponse) {
     enqueueServerToAgentResponse(initialResponse);
 
     callbacks = spy(new TestCallbacks());
-    client = OpampClientImpl.create(requestService, state, callbacks);
+    client = OpampClientImpl.create(requestService, state, x -> callbacks);
 
     return takeRequest();
   }