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.genesys.server.component.firehose;

import java.util.ArrayList;
import java.util.Map;
import java.util.Objects;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.transaction.event.TransactionPhase;
import org.springframework.transaction.event.TransactionalApplicationListener;
import org.springframework.transaction.event.TransactionalEventListener;

/**
 *
 * The FirehoseEventListener is a {@link TransactionalApplicationListener} that
 * is attached to the {@link TransactionPhase} {@code TransactionPhase.AFTER_COMMIT}
 *
 * The event listener then updates the Firehose delay queues accordingly.
 *
 * @author Artem Hrybeniuk
 */
// Instantiated in FirehoseConfig! @Component("firehoseEventListener")
public class FirehoseEventListener {

	public static final Logger LOG = LoggerFactory.getLogger(FirehoseEventListener.class);

	@Resource(name = "updatedEventSet")
	private Map<String, FirehoseEvent> updatedEvents;

	@Resource(name = "removedEventSet")
	private Map<String, FirehoseEvent> removedEvents;

	@Resource(name = "createdEventSet")
	private Map<String, FirehoseEvent> createdEvents;

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

		if (FirehoseEvent.EventType.DELETE == eventType) {
			// Remove given FirehoseEvent from CreatedEvents and UpdatedEvents
			createdEvents.remove(firehoseEvent.getKey());
			updatedEvents.remove(firehoseEvent.getKey());
			removedEvents.put(firehoseEvent.getKey(), firehoseEvent);
		} else if (FirehoseEvent.EventType.CREATE.equals(eventType)) {
			createdEvents.put(firehoseEvent.getKey(), firehoseEvent);
		} else if (FirehoseEvent.EventType.UPDATE.equals(eventType)) {
			putUpdatedEvent(firehoseEvent);
		}
	}


	@TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
	public void handleDeleteAllEvent(FirehoseDeleteAllEvent firehoseDeleteAllEvent) {
		// Remove from created and updated only
		var deletedType = firehoseDeleteAllEvent.getClazz();

		// hazelcastMap.entrySet() is immutable!
		// Unsupported: createdEvents.entrySet().removeIf(entry-> Objects.equals(entry.getValue().getClazz(), deletedType));
		for (var event : new ArrayList<>(createdEvents.values())) {
			if (Objects.equals(event.getClazz(), deletedType)) {
				createdEvents.remove(event.getKey());
			}
		}
		// Unsupported: updatedEvents.entrySet().removeIf(entry -> Objects.equals(entry.getValue().getClazz(), firehoseDeleteAllEvent.getClazz()));
		for (var event : new ArrayList<>(updatedEvents.values())) {
			if (Objects.equals(event.getClazz(), deletedType)) {
				updatedEvents.remove(event.getKey());
			}
		}
	}

	private void putUpdatedEvent(FirehoseEvent firehoseEvent) {
		if (LOG.isTraceEnabled()) {
			LOG.trace("Replacing {} {}", firehoseEvent.getEventType(), firehoseEvent.getKey());
		}
		// replace equal FirehoseEvent in CreatedEvents if contains
		boolean replaceCreatedEvent = replaceIfContains(createdEvents, firehoseEvent);
		if (! replaceCreatedEvent) {
			// replace/put FirehoseEvent to UpdatedEvents
			updatedEvents.put(firehoseEvent.getKey(), firehoseEvent);
		}
	}

	private boolean replaceIfContains(Map<String, FirehoseEvent> events, FirehoseEvent firehoseEvent) {
		// replace Entity and ModifiedDate for equal FirehoseEvent in given Collection
		return events.replace(firehoseEvent.getKey(), firehoseEvent) != null;
	}
}