format

ralscha · Aug 30, 2023 · e64b3c1 · e64b3c1
1 parent 0de59fd
commit e64b3c1
Show file tree

Hide file tree

Showing 13 changed files with 159 additions and 204 deletions.
diff --git a/src/main/java/ch/rasc/sse/eventbus/DefaultSubscriptionRegistry.java b/src/main/java/ch/rasc/sse/eventbus/DefaultSubscriptionRegistry.java
@@ -50,8 +50,7 @@ public void subscribe(String clientId, String event) {
 	 */
 	@Override
 	public void unsubscribe(String clientId, String event) {
-		this.eventSubscribers.computeIfPresent(event,
-				(k, set) -> set.remove(clientId) && set.isEmpty() ? null : set);
+		this.eventSubscribers.computeIfPresent(event, (k, set) -> set.remove(clientId) && set.isEmpty() ? null : set);
 	}
 
 	/*

diff --git a/src/main/java/ch/rasc/sse/eventbus/JacksonDataObjectConverter.java b/src/main/java/ch/rasc/sse/eventbus/JacksonDataObjectConverter.java
@@ -39,8 +39,7 @@ public String convert(SseEvent event) {
 					return this.objectMapper.writeValueAsString(event.data());
 				}
 
-				return this.objectMapper.writerWithView(event.jsonView().get())
-						.writeValueAsString(event.data());
+				return this.objectMapper.writerWithView(event.jsonView().get()).writeValueAsString(event.data());
 			}
 			catch (JsonProcessingException e) {
 				throw new RuntimeException(e);

diff --git a/src/main/java/ch/rasc/sse/eventbus/SseEvent.java b/src/main/java/ch/rasc/sse/eventbus/SseEvent.java
@@ -22,8 +22,7 @@
 import org.immutables.value.Value;
 import org.immutables.value.Value.Style.ImplementationVisibility;
 
-@Value.Style(depluralize = true, visibility = ImplementationVisibility.PACKAGE,
-		overshadowImplementation = true)
+@Value.Style(depluralize = true, visibility = ImplementationVisibility.PACKAGE, overshadowImplementation = true)
 @Value.Immutable
 public interface SseEvent {
 

diff --git a/src/main/java/ch/rasc/sse/eventbus/SseEventBus.java b/src/main/java/ch/rasc/sse/eventbus/SseEventBus.java
@@ -52,8 +52,7 @@ public class SseEventBus {
 
 	private final SseEventBusListener listener;
 
-	public SseEventBus(SseEventBusConfigurer configurer,
-										 SubscriptionRegistry subscriptionRegistry) {
+	public SseEventBus(SseEventBusConfigurer configurer, SubscriptionRegistry subscriptionRegistry) {
 
 		this.subscriptionRegistry = subscriptionRegistry;
 
@@ -71,9 +70,9 @@ public SseEventBus(SseEventBusConfigurer configurer,
 		if (this.taskScheduler != null) {
 			this.taskScheduler.submit(this::eventLoop);
 			this.taskScheduler.scheduleWithFixedDelay(this::reScheduleFailedEvents, 0,
-																								configurer.schedulerDelay().toMillis(), TimeUnit.MILLISECONDS);
+					configurer.schedulerDelay().toMillis(), TimeUnit.MILLISECONDS);
 			this.taskScheduler.scheduleWithFixedDelay(this::cleanUpClients, 0,
-																								configurer.clientExpirationJobDelay().toMillis(), TimeUnit.MILLISECONDS);
+					configurer.clientExpirationJobDelay().toMillis(), TimeUnit.MILLISECONDS);
 		}
 	}
 
@@ -91,17 +90,15 @@ public SseEmitter createSseEmitter(String clientId, String... events) {
 		return createSseEmitter(clientId, 180_000L, false, false, events);
 	}
 
-	public SseEmitter createSseEmitter(String clientId, boolean unsubscribe,
-																		 String... events) {
+	public SseEmitter createSseEmitter(String clientId, boolean unsubscribe, String... events) {
 		return createSseEmitter(clientId, 180_000L, unsubscribe, false, events);
 	}
 
 	public SseEmitter createSseEmitter(String clientId, Long timeout, String... events) {
 		return createSseEmitter(clientId, timeout, false, false, events);
 	}
 
-	public SseEmitter createSseEmitter(String clientId, Long timeout, boolean unsubscribe,
-																		 String... events) {
+	public SseEmitter createSseEmitter(String clientId, Long timeout, boolean unsubscribe, String... events) {
 		return createSseEmitter(clientId, timeout, unsubscribe, false, events);
 	}
 
@@ -115,8 +112,8 @@ public SseEmitter createSseEmitter(String clientId, Long timeout, boolean unsubs
 	 * @param events events the client wants to subscribe
 	 * @return a new SseEmitter instance
 	 */
-	public SseEmitter createSseEmitter(String clientId, Long timeout, boolean unsubscribe,
-																		 boolean completeAfterMessage, String... events) {
+	public SseEmitter createSseEmitter(String clientId, Long timeout, boolean unsubscribe, boolean completeAfterMessage,
+			String... events) {
 		SseEmitter emitter = new SseEmitter(timeout);
 		emitter.onTimeout(emitter::complete);
 		registerClient(clientId, emitter, completeAfterMessage);
@@ -137,12 +134,10 @@ public void registerClient(String clientId, SseEmitter emitter) {
 		this.registerClient(clientId, emitter, false);
 	}
 
-	public void registerClient(String clientId, SseEmitter emitter,
-														 boolean completeAfterMessage) {
+	public void registerClient(String clientId, SseEmitter emitter, boolean completeAfterMessage) {
 		Client client = this.clients.get(clientId);
 		if (client == null) {
-			this.clients.put(clientId,
-											 new Client(clientId, emitter, completeAfterMessage));
+			this.clients.put(clientId, new Client(clientId, emitter, completeAfterMessage));
 		}
 		else {
 			client.updateEmitter(emitter);
@@ -209,21 +204,17 @@ public void handleEvent(SseEvent event) {
 			if (event.clientIds().isEmpty()) {
 				for (Client client : this.clients.values()) {
 					if (!event.excludeClientIds().contains(client.getId())
-							&& this.subscriptionRegistry.isClientSubscribedToEvent(
-							client.getId(), event.event())) {
-						ClientEvent clientEvent = new ClientEvent(client, event,
-																											convertedValue);
+							&& this.subscriptionRegistry.isClientSubscribedToEvent(client.getId(), event.event())) {
+						ClientEvent clientEvent = new ClientEvent(client, event, convertedValue);
 						this.sendQueue.put(clientEvent);
 						this.listener.afterEventQueued(clientEvent, true);
 					}
 				}
 			}
 			else {
 				for (String clientId : event.clientIds()) {
-					if (this.subscriptionRegistry.isClientSubscribedToEvent(clientId,
-																																	event.event())) {
-						ClientEvent clientEvent = new ClientEvent(
-								this.clients.get(clientId), event, convertedValue);
+					if (this.subscriptionRegistry.isClientSubscribedToEvent(clientId, event.event())) {
+						ClientEvent clientEvent = new ClientEvent(this.clients.get(clientId), event, convertedValue);
 						this.sendQueue.put(clientEvent);
 						this.listener.afterEventQueued(clientEvent, true);
 					}
@@ -241,25 +232,22 @@ private void reScheduleFailedEvents() {
 			this.errorQueue.drainTo(failedEvents);
 
 			for (ClientEvent sseClientEvent : failedEvents) {
-				if (this.subscriptionRegistry.isClientSubscribedToEvent(
-						sseClientEvent.getClient().getId(),
+				if (this.subscriptionRegistry.isClientSubscribedToEvent(sseClientEvent.getClient().getId(),
 						sseClientEvent.getSseEvent().event())) {
 					try {
 						this.sendQueue.put(sseClientEvent);
 						try {
 							this.listener.afterEventQueued(sseClientEvent, false);
 						}
 						catch (Exception e) {
-							LogFactory.getLog(SseEventBus.class)
-									.error("calling afterEventQueued hook failed", e);
+							LogFactory.getLog(SseEventBus.class).error("calling afterEventQueued hook failed", e);
 						}
 					}
 					catch (InterruptedException ie) {
 						throw new RuntimeException(ie);
 					}
 					catch (Exception e) {
-						LogFactory.getLog(SseEventBus.class)
-								.error("re-adding event into send queue failed", e);
+						LogFactory.getLog(SseEventBus.class).error("re-adding event into send queue failed", e);
 						try {
 							this.errorQueue.put(sseClientEvent);
 						}
@@ -271,8 +259,7 @@ private void reScheduleFailedEvents() {
 			}
 		}
 		catch (Exception e) {
-			LogFactory.getLog(SseEventBus.class).error("reScheduleFailedEvents failed",
-																								 e);
+			LogFactory.getLog(SseEventBus.class).error("reScheduleFailedEvents failed", e);
 		}
 	}
 
@@ -289,8 +276,7 @@ private void eventLoop() {
 							this.listener.afterEventSent(clientEvent, null);
 						}
 						catch (Exception ex) {
-							LogFactory.getLog(SseEventBus.class)
-									.error("calling afterEventSent hook failed", ex);
+							LogFactory.getLog(SseEventBus.class).error("calling afterEventSent hook failed", ex);
 						}
 					}
 					else {
@@ -305,21 +291,18 @@ private void eventLoop() {
 							this.listener.afterEventSent(clientEvent, e);
 						}
 						catch (Exception ex) {
-							LogFactory.getLog(SseEventBus.class)
-									.error("calling afterEventSent hook failed", ex);
+							LogFactory.getLog(SseEventBus.class).error("calling afterEventSent hook failed", ex);
 						}
 					}
 				}
 				else {
 					String clientId = clientEvent.getClient().getId();
 					this.unregisterClient(clientId);
 					try {
-						this.listener.afterClientsUnregistered(
-								Collections.singleton(clientId));
+						this.listener.afterClientsUnregistered(Collections.singleton(clientId));
 					}
 					catch (Exception ex) {
-						LogFactory.getLog(SseEventBus.class).error(
-								"calling afterClientsUnregistered hook failed", ex);
+						LogFactory.getLog(SseEventBus.class).error("calling afterClientsUnregistered hook failed", ex);
 					}
 				}
 			}
@@ -360,8 +343,7 @@ private String convertObject(SseEvent event) {
 
 	private void cleanUpClients() {
 		if (!this.clients.isEmpty()) {
-			long expirationTime = System.currentTimeMillis()
-					- this.clientExpiration.toMillis();
+			long expirationTime = System.currentTimeMillis() - this.clientExpiration.toMillis();
 			Iterator<Entry<String, Client>> it = this.clients.entrySet().iterator();
 			Set<String> staleClients = new HashSet<>();
 			while (it.hasNext()) {

diff --git a/src/main/java/ch/rasc/sse/eventbus/SseEventBusListener.java b/src/main/java/ch/rasc/sse/eventbus/SseEventBusListener.java
@@ -50,4 +50,5 @@ default void afterEventSent(ClientEvent clientEvent, Exception exception) {
 	default void afterClientsUnregistered(Set<String> clientIds) {
 		// no default implementation
 	}
+
 }
diff --git a/src/main/java/ch/rasc/sse/eventbus/config/SseEventBusConfigurer.java b/src/main/java/ch/rasc/sse/eventbus/config/SseEventBusConfigurer.java
@@ -86,7 +86,9 @@ default BlockingQueue<ClientEvent> sendQueue() {
 		return new LinkedBlockingQueue<>();
 	}
 
-	default ConcurrentMap<String, Client> clients() {return new ConcurrentHashMap<>();}
+	default ConcurrentMap<String, Client> clients() {
+		return new ConcurrentHashMap<>();
+	}
 
 	default SseEventBusListener listener() {
 		return new SseEventBusListener() {