Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Tethering join position #1672

Merged
merged 9 commits into from
Oct 23, 2024
Merged

Tethering join position #1672

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
ebfcbe2
[Java] Fix java tethering join position
marc-adaptive Oct 19, 2024
8c83062
[C] Align c with java publication image tethering join position
marc-adaptive Oct 19, 2024
2d63e97
[Java] Fix join position of tethered subscriptions
marc-adaptive Oct 19, 2024
37b6d59
[C] Align C tethering with java
marc-adaptive Oct 19, 2024
446d613
Merge branch 'refs/heads/master' into tethering-join-position
marc-adaptive Oct 22, 2024
21172a9
Merge branch 'refs/heads/master' into tethering-join-position
marc-adaptive Oct 22, 2024
83b66cb
Merge branch 'refs/heads/master' into tethering-join-position
marc-adaptive Oct 23, 2024
62c6822
[Java] Add position check to untethered system test
marc-adaptive Oct 23, 2024
ccc25b6
[Java] checkstyle
marc-adaptive Oct 23, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion aeron-driver/src/main/c/aeron_ipc_publication.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -372,7 +372,8 @@ void aeron_ipc_publication_check_untethered_subscriptions(
case AERON_SUBSCRIPTION_TETHER_RESTING:
if (now_ns > (tetherable_position->time_of_last_update_ns + resting_timeout_ns))
{
aeron_counter_set_ordered(tetherable_position->value_addr, consumer_position);
int64_t join_position = aeron_ipc_publication_join_position(publication);
aeron_counter_set_ordered(tetherable_position->value_addr, join_position);
aeron_driver_conductor_on_available_image(
conductor,
publication->conductor_fields.managed_resource.registration_id,
Expand Down
3 changes: 2 additions & 1 deletion aeron-driver/src/main/c/aeron_publication_image.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1048,7 +1048,8 @@ void aeron_publication_image_check_untethered_subscriptions(
case AERON_SUBSCRIPTION_TETHER_RESTING:
if (now_ns > (tetherable_position->time_of_last_update_ns + resting_timeout_ns))
{
aeron_counter_set_ordered(tetherable_position->value_addr, *image->rcv_pos_position.value_addr);
int64_t join_position = aeron_publication_image_join_position(image);
aeron_counter_set_ordered(tetherable_position->value_addr, join_position);
aeron_driver_conductor_on_available_image(
conductor,
image->conductor_fields.managed_resource.registration_id,
Expand Down
3 changes: 2 additions & 1 deletion aeron-driver/src/main/java/io/aeron/driver/IpcPublication.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -455,13 +455,14 @@ else if (UntetheredSubscription.State.RESTING == untethered.state)
{
if ((untethered.timeOfLastUpdateNs + untetheredRestingTimeoutNs) - nowNs <= 0)
{
final long joinPosition = joinPosition();
subscriberPositions = ArrayUtil.add(subscriberPositions, untethered.position);
conductor.notifyAvailableImageLink(
registrationId,
sessionId,
untethered.subscriptionLink,
untethered.position.id(),
joinPosition(),
joinPosition,
rawLog.fileName(),
CommonContext.IPC_CHANNEL);
untethered.state(UntetheredSubscription.State.ACTIVE, nowNs, streamId, sessionId);
Expand Down
3 changes: 2 additions & 1 deletion aeron-driver/src/main/java/io/aeron/driver/PublicationImage.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1101,13 +1101,14 @@ else if (UntetheredSubscription.State.RESTING == untethered.state)
{
if ((untethered.timeOfLastUpdateNs + untetheredRestingTimeoutNs) - nowNs <= 0)
{
final long joinPosition = joinPosition();
subscriberPositions = ArrayUtil.add(subscriberPositions, untethered.position);
conductor.notifyAvailableImageLink(
correlationId,
sessionId,
untethered.subscriptionLink,
untethered.position.id(),
joinPosition(),
joinPosition,
rawLog.fileName(),
sourceIdentity);
untethered.state(UntetheredSubscription.State.ACTIVE, nowNs, streamId, sessionId);
Expand Down
2 changes: 1 addition & 1 deletion aeron-driver/src/test/java/io/aeron/driver/UntetheredSubscriptionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,7 +113,7 @@ void shouldLifeCycleTimeoutsAndRelink()
eq(SESSION_ID),
eq(untetheredLink),
anyInt(),
eq(ipcPublication.joinPosition()),
eq(tetheredPosition.get()),
eq(rawLog.fileName()),
eq(CommonContext.IPC_CHANNEL));
}
Expand Down
7 changes: 7 additions & 0 deletions aeron-system-tests/src/test/java/io/aeron/UntetheredSubscriptionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@
import java.util.concurrent.atomic.AtomicInteger;

import static java.util.Arrays.asList;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;

Expand Down Expand Up @@ -218,6 +219,12 @@ void shouldRejoinAfterResting(final String channel)
aeron.conductorAgentInvoker().invoke();
}

if (!channel.startsWith("aeron-spy"))
{
final long tetheredPosition = tetheredSub.imageAtIndex(0).position();
final long untetheredJoinPosition = untetheredSub.imageAtIndex(0).joinPosition();
assertEquals(tetheredPosition, untetheredJoinPosition);
}
return;
}
}
Expand Down