Package io.a2a.extras.queuemanager.replicated.core

Class ReplicatedQueueManager

java.lang.Object

io.a2a.extras.queuemanager.replicated.core.ReplicatedQueueManager

All Implemented Interfaces:

QueueManager

@ApplicationScoped
@Alternative
@Priority(50)
public class ReplicatedQueueManager
extends Object
implements QueueManager

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	ReplicatedQueueManager()	No-args constructor for CDI proxy creation.
	ReplicatedQueueManager(ReplicationStrategy replicationStrategy, TaskStateProvider taskStateProvider, MainEventBus mainEventBus)

Method Summary

Modifier and Type	Method	Description
void	add(String taskId, EventQueue queue)
void	awaitQueuePollerStart(EventQueue eventQueue)
void	close(String taskId)
EventQueue	createOrTap(String taskId)
EventQueue	get(String taskId)
int	getActiveChildQueueCount(String taskId)
EventQueue.EventQueueBuilder	getEventQueueBuilder(String taskId)
void	onReplicatedEvent(ReplicatedEventQueueItem replicatedEvent)
void	onTaskFinalized(TaskFinalizedEvent event)	Observes task finalization events fired AFTER database transaction commits.
EventQueue	tap(String taskId)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface io.a2a.server.events.QueueManager

createBaseEventQueueBuilder

Constructor Details

ReplicatedQueueManager

protected ReplicatedQueueManager()

No-args constructor for CDI proxy creation. CDI requires a non-private constructor to create proxies for @ApplicationScoped beans. All fields are initialized by the @Inject constructor during actual bean creation.

ReplicatedQueueManager

@Inject
public ReplicatedQueueManager(ReplicationStrategy replicationStrategy,
 TaskStateProvider taskStateProvider,
 MainEventBus mainEventBus)

Method Details

add

public void add(String taskId,
 EventQueue queue)

Specified by:

add in interface QueueManager

get

public EventQueue get(String taskId)

Specified by:

get in interface QueueManager

tap

public EventQueue tap(String taskId)

Specified by:

tap in interface QueueManager

close

public void close(String taskId)

Specified by:

close in interface QueueManager

createOrTap

public EventQueue createOrTap(String taskId)

Specified by:

createOrTap in interface QueueManager

awaitQueuePollerStart

public void awaitQueuePollerStart(EventQueue eventQueue)
                           throws InterruptedException

Specified by:

awaitQueuePollerStart in interface QueueManager

Throws:

InterruptedException

onReplicatedEvent

public void onReplicatedEvent(@Observes
 ReplicatedEventQueueItem replicatedEvent)

onTaskFinalized

public void onTaskFinalized(@Observes(during=AFTER_SUCCESS)
 TaskFinalizedEvent event)

Observes task finalization events fired AFTER database transaction commits. This guarantees the task's final state is durably stored before replication. Sends TaskStatusUpdateEvent (not full Task) FIRST, then the poison pill (QueueClosedEvent), ensuring correct event ordering across instances and eliminating race conditions where the poison pill arrives before the final task state. IMPORTANT: We send TaskStatusUpdateEvent instead of full Task to maintain consistency with local event distribution. Clients expect TaskStatusUpdateEvent for status changes, and sending the full Task causes issues in remote instances where clients don't handle bare Task objects the same way they handle TaskStatusUpdateEvent.

Parameters:

event - the task finalized event containing the task ID and final Task

getEventQueueBuilder

public EventQueue.EventQueueBuilder getEventQueueBuilder(String taskId)

Specified by:

getEventQueueBuilder in interface QueueManager

getActiveChildQueueCount

public int getActiveChildQueueCount(String taskId)

Specified by:

getActiveChildQueueCount in interface QueueManager