FirehoseEventListener.java

/*
 * Copyright 2021 Global Crop Diversity Trust
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.gringlobal.component.firehose;

import java.util.Objects;
import java.util.Set;

import javax.annotation.Resource;

import org.gringlobal.component.firehose.FirehoseEvent.EventType;
import org.springframework.stereotype.Component;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalApplicationListener;
import org.springframework.transaction.event.TransactionalEventListener;

import lombok.extern.slf4j.Slf4j;

/**
 *
 * The FirehoseEventListener is a {@link TransactionalApplicationListener} that
 * is attached to the {@link TransactionPhase#AFTER_COMMIT}.
 *
 * The event listener then updates the Firehose event queues.
 * <ol>
 * <li>{@link EventType#CREATE}: adds an entry to the {@code createdEvents} queue.</li>
 * <li>{@link EventType#UPDATE}: if there is a matching record in {@code createdEvents} then update it and set the new timestamp, but keep it as {@link EventType#CREATE}. Otherwise record it in {@code updatedEvents}.</li>
 * <li>{@link EventType#DELETE}: remove matching entries from both {@code createdEvents} and {@code updatedEvents} queues to prevent them being processed as such. The event is added to {@code removedEvents}.</li>
 * </ol>
 */
@Component
@Slf4j
public class FirehoseEventListener {

	@Resource(name = "updatedEventSet")
	private Set<FirehoseEvent> updatedEvents;

	@Resource(name = "removedEventSet")
	private Set<FirehoseEvent> removedEvents;

	@Resource(name = "createdEventSet")
	private Set<FirehoseEvent> createdEvents;

	@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
	public void handleEvent(FirehoseEvent firehoseEvent) {
		EventType eventType = firehoseEvent.getEventType();

		if (EventType.DELETE == eventType) {
			// Remove given FirehoseEvent from CreatedEvents and UpdatedEvents
			createdEvents.removeIf(firehoseEvent::sameReference);
			updatedEvents.removeIf(firehoseEvent::sameReference);
			removedEvents.add(firehoseEvent);
		} else if (EventType.CREATE == eventType) {
			createdEvents.add(firehoseEvent);
		} else if (EventType.UPDATE == eventType) {
			putUpdatedEvent(firehoseEvent);
		} else {
			log.trace("Ignoring event of type {}", firehoseEvent.getEventType());
		}
	}


	@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
	public void handleDeleteAllEvent(FirehoseDeleteAllEvent firehoseDeleteAllEvent) {
		// Remove from created and updated only
		createdEvents.removeIf(event -> Objects.equals(event.getClazz(), firehoseDeleteAllEvent.getClazz()));
		updatedEvents.removeIf(event -> Objects.equals(event.getClazz(), firehoseDeleteAllEvent.getClazz()));
	}

	private void putUpdatedEvent(FirehoseEvent firehoseEvent) {
		// replace equal FirehoseEvent in CreatedEvents if contains
		boolean replaceCreatedEvent = replaceIfContains(createdEvents, firehoseEvent);
		if (!replaceCreatedEvent) {
			// replace equal FirehoseEvent in UpdatedEvents if contains
			boolean replaceUpdatedEvent = replaceIfContains(updatedEvents, firehoseEvent);
			if (!replaceUpdatedEvent) {
				// add FirehoseEvent to UpdatedEvents
				updatedEvents.add(firehoseEvent);
			}
		}
	}

	private boolean replaceIfContains(Set<FirehoseEvent> events, FirehoseEvent firehoseEvent) {
		// replace Entity and ModifiedDate for equal FirehoseEvent in given Collection
		if (events.removeIf(firehoseEvent::sameReference)) {
			events.add(firehoseEvent);
			return true;
		}
		return false;
	}
}