Queue management

Get a queue instance

A Queue represents a single message queue for a specific TimelineIdentifier within a TimelineStore. Create a TimelineQueue instance using the TimelineStore interface.

TimelineIdentifier identifier = new TimelineIdentifier.Builder()
        .addField("timeline_id", "group_1")
        .build();

// A message queue (Queue) for a specific identifier in a TimelineStore.
TimelineQueue timelineQueue = timelineStore.createTimelineQueue(identifier);

A TimelineQueue instance provides the following operations for a specific identifier in a data store:

• Write: synchronous write (store), asynchronous write (storeAsync), and batch write (batchStore)

• Update: synchronous update (update) and asynchronous update (updateAsync)

• Read: single-message read (get) and range read (scan)

• Delete: single-message delete (delete)

Store

Write a message synchronously. Two methods are available, corresponding to the two ways to implement SequenceId: using an auto-increment column or setting it manually. Configure the implementation in TimelineSchema.

timelineQueue.store(message); // SequenceId implemented by an auto-increment column.
timelineQueue.store(sequenceId, message); // Manually set the SequenceId.

StoreAsync

Write a message asynchronously. Define a TimelineCallback to handle success and failure. The method returns Future<TimelineEntry>.

TimelineCallback callback = new TimelineCallback() {
    @Override
    public void onCompleted(TimelineIdentifier i, TimelineMessage m, TimelineEntry t) {
        // Do something on success.
    }

    @Override
    public void onFailed(TimelineIdentifier i, TimelineMessage m, Exception e) {
        // Do something on failure.
    }
};

timelineQueue.storeAsync(message, callback); // SequenceId implemented by an auto-increment column.
timelineQueue.storeAsync(sequenceId, message, callback); // Manually set the SequenceId.

BatchStore

Write multiple messages in a single batch. Both callback and non-callback variants are supported.

timelineQueue.batchStore(message); // SequenceId implemented by an auto-increment column.
timelineQueue.batchStore(sequenceId, message); // Manually set the SequenceId.

timelineQueue.batchStore(message, callback); // SequenceId implemented by an auto-increment column.
timelineQueue.batchStore(sequenceId, message, callback); // Manually set the SequenceId.

Get

Read a single message by its SequenceId. If the message does not exist, this method does not throw an error and returns null.

timelineQueue.get(sequenceId);

GetLatestTimelineEntry

Read the most recent message. If no message exists, this method does not throw an error and returns null.

timelineQueue.getLatestTimelineEntry();

GetLatestSequenceId

Get the SequenceId of the most recent message. If no message exists, this method does not throw an error and returns 0.

timelineQueue.getLatestSequenceId();

Update

Update a message synchronously by its SequenceId.

TimelineMessage message = new TimelineMessage().setField("text", "Timeline is fine.");

// Update the message with a new field.
message.setField("text", "new value");
timelineQueue.update(sequenceId, message);

UpdateAsync

Update a message asynchronously by its SequenceId. Define a TimelineCallback to handle success and failure. The method returns Future<TimelineEntry>.

TimelineMessage oldMessage = new TimelineMessage().setField("text", "Timeline is fine.");
TimelineCallback callback = new TimelineCallback() {
    @Override
    public void onCompleted(TimelineIdentifier i, TimelineMessage m, TimelineEntry t) {
        // Do something on success.
    }

    @Override
    public void onFailed(TimelineIdentifier i, TimelineMessage m, Exception e) {
        // Do something on failure.
    }
};

TimelineMessage newMessage = oldMessage;
newMessage.setField("text", "new value");
timelineQueue.updateAsync(sequenceId, newMessage, callback);

Delete

Delete a single message by its SequenceId.

timelineQueue.delete(sequenceId);

Scan

Read a range of messages in forward or reverse order using a ScanParameter. Returns Iterator<TimelineEntry> for sequential access.

ScanParameter scanParameter = new ScanParameter().scanBackward(Long.MAX_VALUE, 0);

timelineQueue.scan(scanParameter);