org.eclipse.mdm.api.odsadapter/src/main/java/org/eclipse/mdm/api/odsadapter/notification/peak/PeakNotificationManager.java - mdmbl/org.eclipse.mdm - Git at Google

 /********************************************************************************
  * Copyright (c) 2015-2019 Contributors to the Eclipse Foundation
  *
  * See the NOTICE file(s) distributed with this work for additional
  * information regarding copyright ownership.
  *
  * This program and the accompanying materials are made available under the
  * terms of the Eclipse Public License v. 2.0 which is available at
  * http://www.eclipse.org/legal/epl-2.0.
  *
  * SPDX-License-Identifier: EPL-2.0
  *
  ********************************************************************************/

 package org.eclipse.mdm.api.odsadapter.notification.peak;

 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;

 import javax.ws.rs.client.Client;
 import javax.ws.rs.client.ClientBuilder;
 import javax.ws.rs.client.WebTarget;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response.Status;

 import org.eclipse.mdm.api.base.adapter.EntityType;
 import org.eclipse.mdm.api.base.model.User;
 import org.eclipse.mdm.api.base.notification.NotificationException;
 import org.eclipse.mdm.api.base.notification.NotificationFilter;
 import org.eclipse.mdm.api.base.notification.NotificationListener;
 import org.eclipse.mdm.api.base.notification.NotificationService;
 import org.eclipse.mdm.api.base.query.QueryService;
 import org.eclipse.mdm.api.odsadapter.lookup.config.EntityConfig.Key;
 import org.eclipse.mdm.api.odsadapter.notification.NotificationEntityLoader;
 import org.eclipse.mdm.api.odsadapter.query.ODSModelManager;
 import org.eclipse.mdm.api.odsadapter.utils.ODSUtils;
 import org.glassfish.jersey.media.sse.EventInput;
 import org.glassfish.jersey.media.sse.SseFeature;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.peaksolution.ods.notification.protobuf.NotificationProtos.Notification;

 /**
  * Notification manager for handling notifications from the Peak ODS Server
  * Notification Plugin
  *
  * @since 1.0.0
  * @author Matthias Koller, Peak Solution GmbH
  *
  */
 public class PeakNotificationManager implements NotificationService {

 	private static final Logger LOGGER = LoggerFactory.getLogger(PeakNotificationManager.class);

 	private final Client client;
 	private final WebTarget endpoint;

 	private final Map<String, EventProcessor> processors = new HashMap<>();

 	private final ExecutorService executor = Executors.newCachedThreadPool();

 	private final ODSModelManager modelManager;

 	private final NotificationEntityLoader loader;

 	/**
 	 * @param modelManager
 	 * @param url                    URL of the notification plugin
 	 * @param loadContextDescribable if true, the corresponding context describable
 	 *                               is loaded if a notification for a context root
 	 *                               or context component is received.
 	 * @throws NotificationException Thrown if the manager cannot connect to the
 	 *                               notification server.
 	 */
 	public PeakNotificationManager(ODSModelManager modelManager, QueryService queryService, String url,
 			boolean loadContextDescribable) throws NotificationException {
 		this.modelManager = modelManager;
 		loader = new NotificationEntityLoader(modelManager, queryService, loadContextDescribable);

 		try {
 			client = ClientBuilder.newBuilder().register(SseFeature.class).register(JsonMessageBodyProvider.class)
 					.build();

 			endpoint = client.target(url).path("events");

 			Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> reconnect(), 10, 10, TimeUnit.SECONDS);
 		} catch (Exception e) {
 			throw new NotificationException("Could not create " + PeakNotificationManager.class.getName() + "!", e);
 		}
 	}

 	/*
 	 * (non-Javadoc)
 	 *
 	 * @see org.eclipse.mdm.api.base.notification.NotificationManager#register(java.
 	 * lang.String, org.eclipse.mdm.api.base.notification.NotificationFilter,
 	 * org.eclipse.mdm.api.base.notification.NotificationListener)
 	 */
 	@Override
 	public void register(String registration, NotificationFilter filter, NotificationListener listener)
 			throws NotificationException {
 		LOGGER.info("Starting registration for with name: {}", registration);

 		Response response = endpoint.path(registration).request().post(
 				javax.ws.rs.client.Entity.entity(ProtobufConverter.from(filter), MediaType.APPLICATION_JSON_TYPE));

 		if (response.getStatusInfo().getStatusCode() == Status.CONFLICT.getStatusCode()) {
 			if (LOGGER.isInfoEnabled()) {
 				LOGGER.info("A registration with the name already exists: {}", response.readEntity(String.class));
 				LOGGER.info("Trying to reregister...");
 			}
 			deregister(registration);
 			LOGGER.info("Deregisteration successful.");
 			register(registration, filter, listener);
 			return;
 		}

 		if (response.getStatusInfo().getStatusCode() != Status.OK.getStatusCode()) {
 			throw new NotificationException(
 					"Could not create registration at notification service: " + response.readEntity(String.class));
 		}

 		recreateEventStream(registration, listener);
 	}

 	/*
 	 * (non-Javadoc)
 	 *
 	 * @see
 	 * org.eclipse.mdm.api.base.notification.NotificationManager#deregister(java
 	 * .lang.String)
 	 */
 	@Override
 	public void deregister(String registration) {
 		if (processors.containsKey(registration)) {
 			close(registration);
 		}

 		endpoint.path(registration).request().delete();
 	}

 	@Override
 	public void close(boolean isDeregisterAll) throws NotificationException {
 		LOGGER.info("Closing NotificationManager...");

 		for (String registration : processors.keySet()) {
 			if (isDeregisterAll) {
 				LOGGER.debug("Deregistering '{}'.", registration);
 				deregister(registration);
 			} else {
 				LOGGER.debug("Disconnecting '{}'.", registration);
 				close(registration);
 			}
 		}
 	}

 	/**
 	 * Recreates the {@link EventInput}, attaches an {@link EventProcessor} to it
 	 * and stores the {@link EventProcessor} in the processors map.
 	 *
 	 * @param registration
 	 * @param listener
 	 * @throws NotificationException
 	 */
 	private void recreateEventStream(String registration, NotificationListener listener) throws NotificationException {
 		try {
 			LOGGER.info("Requesting event input for {}", registration);
 			EventInput eventInput = endpoint.path(registration).request(SseFeature.SERVER_SENT_EVENTS_TYPE)
 					.get(EventInput.class);

 			LOGGER.info("Received event input, starting event processor.");
 			EventProcessor processor = new EventProcessor(eventInput, listener, this, MediaType.APPLICATION_JSON_TYPE);

 			executor.submit(processor);

 			processors.put(registration, processor);
 			LOGGER.info("Event processor started.");
 		} catch (Exception e) {
 			try {
 				deregister(registration);
 			} catch (Exception ex) {
 				LOGGER.error("Exception upon deregistering!");
 			}
 			throw new NotificationException("Could not create event input stream!", e);
 		}
 	}

 	/**
 	 * Checks if any registered event processor got disconnected and tries to
 	 * recreate the event stream, if necessary.
 	 */
 	private void reconnect() {

 		for (Map.Entry<String, EventProcessor> entry : processors.entrySet()) {
 			if (entry.getValue().isDisconnected()) {
 				try {
 					LOGGER.trace("Registration '{}' was disconnected and will be recreated.", entry.getKey());
 					recreateEventStream(entry.getKey(), entry.getValue().getListener());
 				} catch (NotificationException e) {
 					LOGGER.warn("Cannot recreate event stream for registration " + entry.getKey(), e);
 				}
 			}
 		}
 	}

 	private void close(String registration) {
 		if (processors.containsKey(registration)) {
 			EventProcessor processor = processors.get(registration);
 			processor.stop();
 			processors.remove(registration);
 		}
 	}

 	/**
 	 * Handler for Exceptions during event processing.
 	 *
 	 * @param e Exception which occured during event processing.
 	 */
 	void processException(Exception e) {
 		LOGGER.error("Exception during notification processing!", e);
 	}

 	/**
 	 * Handler for notifications.
 	 *
 	 * @param n                    notification to process.
 	 * @param notificationListener notification listener for handling the
 	 *                             notification.
 	 */
 	void processNotification(Notification n, NotificationListener notificationListener) {
 		if (LOGGER.isDebugEnabled()) {
 			LOGGER.debug("Processing notification event: " + n);
 		}

 		try {
 			User user = loader.load(new Key<>(User.class), Long.toString(n.getUserId()));

 			EntityType entityType = modelManager.getEntityTypeById(Long.toString(n.getAid()));

 			if (LOGGER.isTraceEnabled()) {
 				LOGGER.trace("Notification event with: entityType=" + entityType + ", user=" + user);
 			}
 			switch (n.getType()) {
 			case NEW:
 				notificationListener.instanceCreated(loader.loadEntities(entityType, n.getIidList().stream()
 						.map(id -> id.toString()).filter(ODSUtils::isValidID).collect(Collectors.toList())), user);
 				break;
 			case MODIFY:
 				notificationListener.instanceModified(loader.loadEntities(entityType, n.getIidList().stream()
 						.map(id -> id.toString()).filter(ODSUtils::isValidID).collect(Collectors.toList())), user);
 				break;
 			case DELETE:
 				notificationListener.instanceDeleted(entityType, n.getIidList().stream().map(id -> id.toString())
 						.filter(ODSUtils::isValidID).collect(Collectors.toList()), user);
 				break;
 			case MODEL:
 				notificationListener.modelModified(entityType, user);
 				break;
 			case SECURITY:
 				notificationListener.securityModified(entityType, n.getIidList().stream().map(id -> id.toString())
 						.filter(ODSUtils::isValidID).collect(Collectors.toList()), user);
 				break;
 			default:
 				processException(new NotificationException("Invalid notification type!"));
 			}
 		} catch (Exception e) {
 			processException(new NotificationException("Could not process notification!", e));
 		}
 	}

 }
	/********************************************************************************
	* Copyright (c) 2015-2019 Contributors to the Eclipse Foundation
	*
	* See the NOTICE file(s) distributed with this work for additional
	* information regarding copyright ownership.
	*
	* This program and the accompanying materials are made available under the
	* terms of the Eclipse Public License v. 2.0 which is available at
	* http://www.eclipse.org/legal/epl-2.0.
	*
	* SPDX-License-Identifier: EPL-2.0
	*
	********************************************************************************/

	package org.eclipse.mdm.api.odsadapter.notification.peak;

	import java.util.HashMap;
	import java.util.Map;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Executors;
	import java.util.concurrent.TimeUnit;
	import java.util.stream.Collectors;

	import javax.ws.rs.client.Client;
	import javax.ws.rs.client.ClientBuilder;
	import javax.ws.rs.client.WebTarget;
	import javax.ws.rs.core.MediaType;
	import javax.ws.rs.core.Response;
	import javax.ws.rs.core.Response.Status;

	import org.eclipse.mdm.api.base.adapter.EntityType;
	import org.eclipse.mdm.api.base.model.User;
	import org.eclipse.mdm.api.base.notification.NotificationException;
	import org.eclipse.mdm.api.base.notification.NotificationFilter;
	import org.eclipse.mdm.api.base.notification.NotificationListener;
	import org.eclipse.mdm.api.base.notification.NotificationService;
	import org.eclipse.mdm.api.base.query.QueryService;
	import org.eclipse.mdm.api.odsadapter.lookup.config.EntityConfig.Key;
	import org.eclipse.mdm.api.odsadapter.notification.NotificationEntityLoader;
	import org.eclipse.mdm.api.odsadapter.query.ODSModelManager;
	import org.eclipse.mdm.api.odsadapter.utils.ODSUtils;
	import org.glassfish.jersey.media.sse.EventInput;
	import org.glassfish.jersey.media.sse.SseFeature;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.peaksolution.ods.notification.protobuf.NotificationProtos.Notification;

	/**
	* Notification manager for handling notifications from the Peak ODS Server
	* Notification Plugin
	*
	* @since 1.0.0
	* @author Matthias Koller, Peak Solution GmbH
	*
	*/
	public class PeakNotificationManager implements NotificationService {

	private static final Logger LOGGER = LoggerFactory.getLogger(PeakNotificationManager.class);

	private final Client client;
	private final WebTarget endpoint;

	private final Map<String, EventProcessor> processors = new HashMap<>();

	private final ExecutorService executor = Executors.newCachedThreadPool();

	private final ODSModelManager modelManager;

	private final NotificationEntityLoader loader;

	/**
	* @param modelManager
	* @param url URL of the notification plugin
	* @param loadContextDescribable if true, the corresponding context describable
	* is loaded if a notification for a context root
	* or context component is received.
	* @throws NotificationException Thrown if the manager cannot connect to the
	* notification server.
	*/
	public PeakNotificationManager(ODSModelManager modelManager, QueryService queryService, String url,
	boolean loadContextDescribable) throws NotificationException {
	this.modelManager = modelManager;
	loader = new NotificationEntityLoader(modelManager, queryService, loadContextDescribable);

	try {
	client = ClientBuilder.newBuilder().register(SseFeature.class).register(JsonMessageBodyProvider.class)
	.build();

	endpoint = client.target(url).path("events");

	Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> reconnect(), 10, 10, TimeUnit.SECONDS);
	} catch (Exception e) {
	throw new NotificationException("Could not create " + PeakNotificationManager.class.getName() + "!", e);
	}
	}

	/*
	* (non-Javadoc)
	*
	* @see org.eclipse.mdm.api.base.notification.NotificationManager#register(java.
	* lang.String, org.eclipse.mdm.api.base.notification.NotificationFilter,
	* org.eclipse.mdm.api.base.notification.NotificationListener)
	*/
	@Override
	public void register(String registration, NotificationFilter filter, NotificationListener listener)
	throws NotificationException {
	LOGGER.info("Starting registration for with name: {}", registration);

	Response response = endpoint.path(registration).request().post(
	javax.ws.rs.client.Entity.entity(ProtobufConverter.from(filter), MediaType.APPLICATION_JSON_TYPE));

	if (response.getStatusInfo().getStatusCode() == Status.CONFLICT.getStatusCode()) {
	if (LOGGER.isInfoEnabled()) {
	LOGGER.info("A registration with the name already exists: {}", response.readEntity(String.class));
	LOGGER.info("Trying to reregister...");
	}
	deregister(registration);
	LOGGER.info("Deregisteration successful.");
	register(registration, filter, listener);
	return;
	}

	if (response.getStatusInfo().getStatusCode() != Status.OK.getStatusCode()) {
	throw new NotificationException(
	"Could not create registration at notification service: " + response.readEntity(String.class));
	}

	recreateEventStream(registration, listener);
	}

	/*
	* (non-Javadoc)
	*
	* @see
	* org.eclipse.mdm.api.base.notification.NotificationManager#deregister(java
	* .lang.String)
	*/
	@Override
	public void deregister(String registration) {
	if (processors.containsKey(registration)) {
	close(registration);
	}

	endpoint.path(registration).request().delete();
	}

	@Override
	public void close(boolean isDeregisterAll) throws NotificationException {
	LOGGER.info("Closing NotificationManager...");

	for (String registration : processors.keySet()) {
	if (isDeregisterAll) {
	LOGGER.debug("Deregistering '{}'.", registration);
	deregister(registration);
	} else {
	LOGGER.debug("Disconnecting '{}'.", registration);
	close(registration);
	}
	}
	}

	/**
	* Recreates the {@link EventInput}, attaches an {@link EventProcessor} to it
	* and stores the {@link EventProcessor} in the processors map.
	*
	* @param registration
	* @param listener
	* @throws NotificationException
	*/
	private void recreateEventStream(String registration, NotificationListener listener) throws NotificationException {
	try {
	LOGGER.info("Requesting event input for {}", registration);
	EventInput eventInput = endpoint.path(registration).request(SseFeature.SERVER_SENT_EVENTS_TYPE)
	.get(EventInput.class);

	LOGGER.info("Received event input, starting event processor.");
	EventProcessor processor = new EventProcessor(eventInput, listener, this, MediaType.APPLICATION_JSON_TYPE);

	executor.submit(processor);

	processors.put(registration, processor);
	LOGGER.info("Event processor started.");
	} catch (Exception e) {
	try {
	deregister(registration);
	} catch (Exception ex) {
	LOGGER.error("Exception upon deregistering!");
	}
	throw new NotificationException("Could not create event input stream!", e);
	}
	}

	/**
	* Checks if any registered event processor got disconnected and tries to
	* recreate the event stream, if necessary.
	*/
	private void reconnect() {

	for (Map.Entry<String, EventProcessor> entry : processors.entrySet()) {
	if (entry.getValue().isDisconnected()) {
	try {
	LOGGER.trace("Registration '{}' was disconnected and will be recreated.", entry.getKey());
	recreateEventStream(entry.getKey(), entry.getValue().getListener());
	} catch (NotificationException e) {
	LOGGER.warn("Cannot recreate event stream for registration " + entry.getKey(), e);
	}
	}
	}
	}

	private void close(String registration) {
	if (processors.containsKey(registration)) {
	EventProcessor processor = processors.get(registration);
	processor.stop();
	processors.remove(registration);
	}
	}

	/**
	* Handler for Exceptions during event processing.
	*
	* @param e Exception which occured during event processing.
	*/
	void processException(Exception e) {
	LOGGER.error("Exception during notification processing!", e);
	}

	/**
	* Handler for notifications.
	*
	* @param n notification to process.
	* @param notificationListener notification listener for handling the
	* notification.
	*/
	void processNotification(Notification n, NotificationListener notificationListener) {
	if (LOGGER.isDebugEnabled()) {
	LOGGER.debug("Processing notification event: " + n);
	}

	try {
	User user = loader.load(new Key<>(User.class), Long.toString(n.getUserId()));

	EntityType entityType = modelManager.getEntityTypeById(Long.toString(n.getAid()));

	if (LOGGER.isTraceEnabled()) {
	LOGGER.trace("Notification event with: entityType=" + entityType + ", user=" + user);
	}
	switch (n.getType()) {
	case NEW:
	notificationListener.instanceCreated(loader.loadEntities(entityType, n.getIidList().stream()
	.map(id -> id.toString()).filter(ODSUtils::isValidID).collect(Collectors.toList())), user);
	break;
	case MODIFY:
	notificationListener.instanceModified(loader.loadEntities(entityType, n.getIidList().stream()
	.map(id -> id.toString()).filter(ODSUtils::isValidID).collect(Collectors.toList())), user);
	break;
	case DELETE:
	notificationListener.instanceDeleted(entityType, n.getIidList().stream().map(id -> id.toString())
	.filter(ODSUtils::isValidID).collect(Collectors.toList()), user);
	break;
	case MODEL:
	notificationListener.modelModified(entityType, user);
	break;
	case SECURITY:
	notificationListener.securityModified(entityType, n.getIidList().stream().map(id -> id.toString())
	.filter(ODSUtils::isValidID).collect(Collectors.toList()), user);
	break;
	default:
	processException(new NotificationException("Invalid notification type!"));
	}
	} catch (Exception e) {
	processException(new NotificationException("Could not process notification!", e));
	}
	}

	}