diff --git a/continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java b/continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java index 1230cd2a..bb3a1a72 100644 --- a/continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java +++ b/continuum-core-vertx/src/main/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusService.java @@ -56,6 +56,8 @@ import java.util.List; import java.util.Map; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicBoolean; /** * Default implementation of {@link EventBusService} using the vertx {@link io.vertx.core.eventbus.EventBus} as a backend @@ -76,6 +78,15 @@ public class DefaultEventBusService implements EventBusService { private IgniteCache> subscriptionsCache; @Autowired private Vertx vertx; + /** + * Reference counted set of event bus addresses this node hosts a local handler for. + * An entry is added by {@link #_listen} when a consumer's handler is attached on this node and removed + * when that consumer is unregistered, so membership means "a local consumer exists at this address". + * This is local by construction: {@link #_listen} only ever registers consumers on this node, other nodes' + * registrations live solely in the cluster's subscription registry. The send path uses this to prefer + * local delivery for service RPCs and avoid an unnecessary cluster hop. + */ + private final Map localListenerCounts = new ConcurrentHashMap<>(); @PostConstruct public void init(){ @@ -111,7 +122,7 @@ public Mono>> listenWithAck(String cri) { return Mono.create(sink -> { final MessageConsumer consumer = vertx.eventBus().consumer(cri); - final ConnectableFlux> flux = _listen(null, consumer).publish(); + final ConnectableFlux> flux = _listen(cri, consumer).publish(); consumer.completion().onComplete(ar ->{ if(ar.succeeded()){ sink.success(flux); @@ -185,8 +196,9 @@ public Flux monitorListenerStatus(String cri) { @Override public void send(Event event) { - DeliveryOptions deliveryOptions = createDeliveryOptions(event); - vertx.eventBus().send(event.cri().baseResource(), + String baseResource = event.cri().baseResource(); + DeliveryOptions deliveryOptions = createDeliveryOptions(event, baseResource); + vertx.eventBus().send(baseResource, event.data(), deliveryOptions); } @@ -195,10 +207,11 @@ public void send(Event event) { public Mono sendWithAck(Event event) { Validate.notNull(event, "Event must not be null"); return Mono.create(sink -> { - DeliveryOptions deliveryOptions = createDeliveryOptions(event); + String baseResource = event.cri().baseResource(); + DeliveryOptions deliveryOptions = createDeliveryOptions(event, baseResource); // We expect that a response will be sent upon receipt. This will happen automatically if the listener is created with this class. vertx.eventBus() - .request(event.cri().baseResource(), + .request(baseResource, event.data(), deliveryOptions) .onComplete(reply -> { @@ -220,8 +233,20 @@ private Flux> _listen(String cri, MessageConsumer vertxEve } Flux> ret = Flux.create(fluxSink -> { + // Record that this node now hosts a local handler for cri. Done here, where the consumer's handler is + // attached, and undone in onDispose, where the consumer is unregistered, so the count is coupled to the + // consumer's actual registration lifecycle on this node. + localListenerCounts.merge(cri, 1, Integer::sum); + final AtomicBoolean unregistered = new AtomicBoolean(false); + // Setup all required handlers that are needed prior to consuming messages - fluxSink.onDispose(consumer::unregister); + fluxSink.onDispose(() -> { + // Decrement exactly once on the first unregister, removing the entry when no local handlers remain. + if(unregistered.compareAndSet(false, true)){ + localListenerCounts.computeIfPresent(cri, (k, n) -> n == 1 ? null : n - 1); + } + consumer.unregister(); + }); // TODO: deal with back pressure properly.. ? //fluxSink.onRequest() @@ -245,7 +270,7 @@ private Flux> _listen(String cri, MessageConsumer vertxEve return ret; // ensure message delivery happens on vertx event loop, not sure but this by itself did not move the next above to the work loop } - private DeliveryOptions createDeliveryOptions(Event event){ + private DeliveryOptions createDeliveryOptions(Event event, String baseResource){ DeliveryOptions deliveryOptions = new DeliveryOptions(); deliveryOptions.setTracingPolicy(TracingPolicy.IGNORE); // fast path for MultiMapMetadataAdapter's @@ -257,7 +282,29 @@ private DeliveryOptions createDeliveryOptions(Event event){ } } deliveryOptions.addHeader(EventConstants.CRI_HEADER, event.cri().raw()); + + // Prefer local delivery when this node already hosts a handler for the target service address. + // In the clustered event bus the point-to-point selector round-robins sends across every node + // registered for the address, so without this ~half of a node's own service calls get shipped to a + // remote node. localOnly=true bypasses the cluster selector and delivers to the local handler. + // Only set when this node actually hosts a handler, otherwise localOnly would fail with NO_HANDLERS. + if(shouldPreferLocalDelivery(event, baseResource)){ + deliveryOptions.setLocalOnly(true); + } + return deliveryOptions; } + /** + * Determines if a send to the given {@code baseResource} should be delivered to a local handler only. + * True only for point-to-point {@link EventConstants#SERVICE_DESTINATION_SCHEME service} sends whose + * address is currently hosted by a local handler on this node. This is a lock free + * {@link Map#containsKey} plus a scheme check, no cluster or registry query. + * Package private for testing. + */ + boolean shouldPreferLocalDelivery(Event event, String baseResource){ + return EventConstants.SERVICE_DESTINATION_SCHEME.equals(event.cri().scheme()) + && localListenerCounts.containsKey(baseResource); + } + } diff --git a/continuum-core-vertx/src/test/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusServiceLocalDeliveryTest.java b/continuum-core-vertx/src/test/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusServiceLocalDeliveryTest.java new file mode 100644 index 00000000..325499c0 --- /dev/null +++ b/continuum-core-vertx/src/test/java/org/kinotic/continuum/internal/core/api/event/DefaultEventBusServiceLocalDeliveryTest.java @@ -0,0 +1,144 @@ +/* + * + * Copyright 2008-2021 Kinotic and the original author or authors. + * + * 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 + * + * https://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.kinotic.continuum.internal.core.api.event; + +import io.vertx.core.Vertx; +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.kinotic.continuum.core.api.event.Event; +import org.springframework.test.util.ReflectionTestUtils; +import reactor.core.Disposable; + +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Verifies the {@code localOnly} delivery predicate of {@link DefaultEventBusService}: when this node hosts a + * local handler for a service address, sends to that address should prefer local delivery, otherwise they should + * fall back to normal cluster routing. + * + * Uses a real (non clustered) {@link Vertx} so the consumer registration / unregistration lifecycle that drives + * the predicate is exercised end to end rather than mocked. + */ +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +public class DefaultEventBusServiceLocalDeliveryTest { + + private Vertx vertx; + private DefaultEventBusService eventBusService; + + @BeforeAll + public void startVertx() { + vertx = Vertx.vertx(); + } + + @AfterAll + public void stopVertx() { + if (vertx != null) { + vertx.close(); + } + } + + @BeforeEach + public void setUp() { + // a fresh service per test so the tracked local listener state starts empty + eventBusService = new DefaultEventBusService(); + ReflectionTestUtils.setField(eventBusService, "vertx", vertx); + eventBusService.init(); + } + + @AfterEach + public void tearDown() { + eventBusService = null; + } + + private static Event event(String cri) { + return Event.create(cri, "data".getBytes(StandardCharsets.UTF_8)); + } + + @Test + public void prefersLocalDeliveryForLocallyHostedServiceAddress() { + String baseResource = "srv://org.kinotic.continuum.tests.LocalDeliveryService"; + Disposable subscription = eventBusService.listen(baseResource).subscribe(); + try { + // a send targets the base resource of the request cri (scheme + optional scope + resource name) + Event request = event(baseResource + "/someMethod"); + assertThat(request.cri().baseResource()) + .as("the send address must equal the listened base resource") + .isEqualTo(baseResource); + + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("a service address hosted by a local handler should prefer local delivery") + .isTrue(); + } finally { + subscription.dispose(); + } + } + + @Test + public void doesNotPreferLocalDeliveryForAddressNotHostedLocally() { + String hosted = "srv://org.kinotic.continuum.tests.HostedService"; + Disposable subscription = eventBusService.listen(hosted).subscribe(); + try { + Event request = event("srv://org.kinotic.continuum.tests.OtherService/someMethod"); + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("a service address not hosted locally should fall back to cluster routing") + .isFalse(); + } finally { + subscription.dispose(); + } + } + + @Test + public void doesNotPreferLocalDeliveryForNonServiceScheme() { + // This node hosts a local handler at this exact address, but the non service scheme must still + // suppress the optimization so reply / stream / other sends are left untouched. + String streamAddress = "stream://org.kinotic.continuum.tests.SomeStream"; + Disposable subscription = eventBusService.listen(streamAddress).subscribe(); + try { + Event request = event(streamAddress + "/someMethod"); + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("a non service scheme should not prefer local delivery even when hosted locally") + .isFalse(); + } finally { + subscription.dispose(); + } + } + + @Test + public void clearsAfterConsumerUnregistered() { + String baseResource = "srv://org.kinotic.continuum.tests.EphemeralService"; + Disposable subscription = eventBusService.listen(baseResource).subscribe(); + Event request = event(baseResource + "/someMethod"); + + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("local delivery should be preferred while the consumer is registered") + .isTrue(); + + // unregistering the only local handler must clear the preference so a later send falls back to the cluster + subscription.dispose(); + + assertThat(eventBusService.shouldPreferLocalDelivery(request, request.cri().baseResource())) + .as("local delivery preference should clear once the consumer is unregistered") + .isFalse(); + } +}