feat: switch to upstream processors + refactor (#174)

Author: wolf4ood

core/negotiation-manager/build.gradle.kts …tract-negotiation-tasks/build.gradle.ktscore/negotiation-manager/build.gradle.kts renamed to core/contract-negotiation-tasks/build.gradle.kts core/negotiation-manager/build.gradle.kts renamed to core/contract-negotiation-tasks/build.gradle.kts

@@ -17,12 +17,13 @@ plugins {
 }
 
 dependencies {
-    api(project(":spi:v-core-spi"))
     api(project(":spi:v-task-spi"))
     api(libs.edc.spi.core)
     api(libs.edc.spi.transaction)
     api(libs.edc.spi.protocol)
     api(libs.edc.spi.contract)
     implementation(libs.opentelemetry.instrumentation.annotations)
+    testImplementation(libs.edc.core.controlplane.contract)
+    testImplementation(libs.edc.lib.statemachine)
 }
 

…egotiation/ContractManagerExtension.java …on/ContractNegotiationTaskExtension.javacore/negotiation-manager/src/main/java/org/eclipse/edc/virtual/controlplane/contract/negotiation/ContractManagerExtension.java renamed to core/contract-negotiation-tasks/src/main/java/org/eclipse/edc/virtual/controlplane/contract/negotiation/ContractNegotiationTaskExtension.java core/negotiation-manager/src/main/java/org/eclipse/edc/virtual/controlplane/contract/negotiation/ContractManagerExtension.java renamed to core/contract-negotiation-tasks/src/main/java/org/eclipse/edc/virtual/controlplane/contract/negotiation/ContractNegotiationTaskExtension.java

@@ -14,31 +14,30 @@
 
 package org.eclipse.edc.virtual.controlplane.contract.negotiation;
 
-import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.ConsumerContractNegotiationManager;
 import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.ContractNegotiationPendingGuard;
+import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.NegotiationProcessors;
 import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.observe.ContractNegotiationObservable;
 import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.store.ContractNegotiationStore;
-import org.eclipse.edc.participantcontext.spi.identity.ParticipantIdentityResolver;
 import org.eclipse.edc.runtime.metamodel.annotation.Extension;
 import org.eclipse.edc.runtime.metamodel.annotation.Inject;
 import org.eclipse.edc.runtime.metamodel.annotation.Provider;
-import org.eclipse.edc.spi.message.RemoteMessageDispatcherRegistry;
 import org.eclipse.edc.spi.monitor.Monitor;
 import org.eclipse.edc.spi.system.ServiceExtension;
+import org.eclipse.edc.spi.system.ServiceExtensionContext;
 import org.eclipse.edc.transaction.spi.TransactionContext;
+import org.eclipse.edc.virtual.controlplane.contract.negotiation.listener.ContractNegotiationStateListener;
 import org.eclipse.edc.virtual.controlplane.contract.negotiation.task.ContractNegotiationTaskExecutorImpl;
 import org.eclipse.edc.virtual.controlplane.contract.spi.negotiation.ContractNegotiationTaskExecutor;
-import org.eclipse.edc.virtual.controlplane.participantcontext.spi.ParticipantWebhookResolver;
 import org.eclipse.edc.virtual.controlplane.tasks.TaskService;
 
 import java.time.Clock;
 
-import static org.eclipse.edc.virtual.controlplane.contract.negotiation.ContractManagerExtension.NAME;
+import static org.eclipse.edc.virtual.controlplane.contract.negotiation.ContractNegotiationTaskExtension.NAME;
 
 @Extension(NAME)
-public class ContractManagerExtension implements ServiceExtension {
+public class ContractNegotiationTaskExtension implements ServiceExtension {
 
-    public static final String NAME = "EDC-V Contract Manager";
+    public static final String NAME = "Contract Negotiation Task Extension";
     @Inject
     private ContractNegotiationStore contractNegotiationStore;
 
@@ -52,40 +51,32 @@ public class ContractManagerExtension implements ServiceExtension {
     private Monitor monitor;
 
     @Inject
-    private ParticipantWebhookResolver webhookResolver;
-
-    @Inject
-    private ParticipantIdentityResolver identityResolver;
+    private TaskService taskService;
 
     @Inject
-    private RemoteMessageDispatcherRegistry dispatcherRegistry;
+    private NegotiationProcessors negotiationProcessors;
 
     @Inject
-    private ContractNegotiationObservable observable;
+    private Clock clock;
 
     @Inject
-    private TaskService taskService;
+    private ContractNegotiationObservable contractNegotiationObservable;
 
-    @Inject
-    private Clock clock;
 
-    @Provider
-    public ConsumerContractNegotiationManager consumerContractNegotiationManager() {
-        return new VirtualConsumerContractNegotiationManager(contractNegotiationStore, observable, monitor);
+    @Override
+    public void initialize(ServiceExtensionContext context) {
+        contractNegotiationObservable.registerListener(new ContractNegotiationStateListener(taskService, clock));
     }
 
     @Provider
     public ContractNegotiationTaskExecutor contractNegotiationTaskExecutor() {
         return ContractNegotiationTaskExecutorImpl.Builder.newInstance()
                 .taskService(taskService)
+                .negotiationProcessors(negotiationProcessors)
                 .clock(clock)
-                .identityResolver(identityResolver)
-                .webhookResolver(webhookResolver)
-                .dispatcherRegistry(dispatcherRegistry)
                 .store(contractNegotiationStore)
                 .transactionContext(transactionContext)
                 .pendingGuard(pendingGuard)
-                .observable(observable)
                 .monitor(monitor)
                 .build();
     }

…er/ContractNegotiationStateListener.java …er/ContractNegotiationStateListener.javaextensions/control-plane/tasks/listener/tasks-store-listener/src/main/java/org/eclipse/edc/virtual/controlplane/listener/ContractNegotiationStateListener.java renamed to core/contract-negotiation-tasks/src/main/java/org/eclipse/edc/virtual/controlplane/contract/negotiation/listener/ContractNegotiationStateListener.java extensions/control-plane/tasks/listener/tasks-store-listener/src/main/java/org/eclipse/edc/virtual/controlplane/listener/ContractNegotiationStateListener.java renamed to core/contract-negotiation-tasks/src/main/java/org/eclipse/edc/virtual/controlplane/contract/negotiation/listener/ContractNegotiationStateListener.java

@@ -12,7 +12,7 @@
  *
  */
 
-package org.eclipse.edc.virtual.controlplane.listener;
+package org.eclipse.edc.virtual.controlplane.contract.negotiation.listener;
 
 import org.eclipse.edc.connector.controlplane.contract.spi.negotiation.observe.ContractNegotiationListener;
 import org.eclipse.edc.connector.controlplane.contract.spi.types.negotiation.ContractNegotiation;
@@ -22,14 +22,20 @@
 import org.eclipse.edc.virtual.controlplane.contract.spi.negotiation.tasks.RequestNegotiation;
 import org.eclipse.edc.virtual.controlplane.contract.spi.negotiation.tasks.VerifyNegotiation;
 import org.eclipse.edc.virtual.controlplane.tasks.ProcessTaskPayload;
+import org.eclipse.edc.virtual.controlplane.tasks.Task;
+import org.eclipse.edc.virtual.controlplane.tasks.TaskPayload;
 import org.eclipse.edc.virtual.controlplane.tasks.TaskService;
 
 import java.time.Clock;
 
-public class ContractNegotiationStateListener extends StateListener implements ContractNegotiationListener {
+public class ContractNegotiationStateListener implements ContractNegotiationListener {
+
+    private final TaskService taskService;
+    private final Clock clock;
 
     public ContractNegotiationStateListener(TaskService taskService, Clock clock) {
-        super(taskService, clock);
+        this.taskService = taskService;
+        this.clock = clock;
     }
 
     @Override
@@ -91,4 +97,12 @@ protected <T extends ProcessTaskPayload, B extends ProcessTaskPayload.Builder<T,
                 .processType(negotiation.getType().name());
     }
 
+    protected void storeTask(TaskPayload payload) {
+        var task = Task.Builder.newInstance()
+                .at(clock.millis())
+                .payload(payload)
+                .build();
+        taskService.create(task);
+    }
+
 }