From 317af37725055ba8a30a2036234a3d24d3f6ad10 Mon Sep 17 00:00:00 2001 From: mjpt777 Date: Mon, 17 Jul 2017 20:52:07 +0100 Subject: [PATCH] [Java] Make it clear that the recording is using a stripped down version of the channel. --- .../src/main/resources/archiver-codecs-schema.xml | 2 +- .../java/io/aeron/archiver/ArchiveConductor.java | 15 +++++++-------- .../src/main/java/io/aeron/archiver/Catalog.java | 4 ++-- .../archiver/client/ControlResponsePoller.java | 2 +- .../client/RecordingDescriptorConsumer.java | 4 ++-- .../io/aeron/archiver/ArchiverSystemTest.java | 2 +- .../test/java/io/aeron/archiver/CatalogTest.java | 6 +++--- .../archiver/FailControlResponseListener.java | 2 +- 8 files changed, 18 insertions(+), 19 deletions(-) diff --git a/aeron-archiver-codecs/src/main/resources/archiver-codecs-schema.xml b/aeron-archiver-codecs/src/main/resources/archiver-codecs-schema.xml index d34d8bf216..14d96745b4 100644 --- a/aeron-archiver-codecs/src/main/resources/archiver-codecs-schema.xml +++ b/aeron-archiver-codecs/src/main/resources/archiver-codecs-schema.xml @@ -144,7 +144,7 @@ - + diff --git a/aeron-archiver/src/main/java/io/aeron/archiver/ArchiveConductor.java b/aeron-archiver/src/main/java/io/aeron/archiver/ArchiveConductor.java index 1aaa0ca92e..753ca69f9f 100644 --- a/aeron-archiver/src/main/java/io/aeron/archiver/ArchiveConductor.java +++ b/aeron-archiver/src/main/java/io/aeron/archiver/ArchiveConductor.java @@ -182,8 +182,7 @@ void stopRecording( { try { - final String minimalChannel = minimalChannelBuilder(channel).build(); - final String key = makeKey(streamId, minimalChannel); + final String key = makeKey(streamId, strippedChannelBuilder(channel).build()); final Subscription oldSubscription = subscriptionMap.remove(key); if (oldSubscription != null) { @@ -239,13 +238,13 @@ void startRecordingSubscription( return; } - final String minimalChannel = minimalChannelBuilder(originalChannel).build(); - final String key = makeKey(streamId, minimalChannel); + final String strippedChannel = strippedChannelBuilder(originalChannel).build(); + final String key = makeKey(streamId, strippedChannel); final Subscription oldSubscription = subscriptionMap.get(key); if (oldSubscription == null) { final Subscription subscription = aeron.addSubscription( - minimalChannel, + strippedChannel, streamId, image -> startImageRecording(originalChannel, image), null); @@ -361,7 +360,7 @@ Publication newControlPublication(final String channel, final int streamId) final String controlChannel; if (!channel.contains(CommonContext.TERM_LENGTH_PARAM_NAME)) { - controlChannel = minimalChannelBuilder(channel) + controlChannel = strippedChannelBuilder(channel) .termLength(DEFAULT_CONTROL_TERM_LENGTH) .build(); } @@ -378,7 +377,7 @@ private static String makeKey(final int streamId, final String minimalChannel) return streamId + ':' + minimalChannel; } - private ChannelUriStringBuilder minimalChannelBuilder(final String channel) + private ChannelUriStringBuilder strippedChannelBuilder(final String channel) { final ChannelUri channelUri = ChannelUri.parse(channel); channelBuilder @@ -449,7 +448,7 @@ private ExclusivePublication newReplayPublication( final int termId = (int)((fromPosition / termBufferLength) + initialTermId); final int termOffset = (int)(fromPosition % termBufferLength); - final String channel = minimalChannelBuilder(replayChannel) + final String channel = strippedChannelBuilder(replayChannel) .mtu(mtuLength) .termLength(termBufferLength) .initialTermId(initialTermId) diff --git a/aeron-archiver/src/main/java/io/aeron/archiver/Catalog.java b/aeron-archiver/src/main/java/io/aeron/archiver/Catalog.java index 580a8e974c..c6a9721ac4 100644 --- a/aeron-archiver/src/main/java/io/aeron/archiver/Catalog.java +++ b/aeron-archiver/src/main/java/io/aeron/archiver/Catalog.java @@ -262,7 +262,7 @@ static void initDescriptor( final int mtuLength, final int sessionId, final int streamId, - final String channel, + final String strippedChannel, final String originalChannel, final String sourceIdentity) { @@ -278,7 +278,7 @@ static void initDescriptor( .mtuLength(mtuLength) .sessionId(sessionId) .streamId(streamId) - .channel(channel) + .strippedChannel(strippedChannel) .originalChannel(originalChannel) .sourceIdentity(sourceIdentity); } diff --git a/aeron-archiver/src/main/java/io/aeron/archiver/client/ControlResponsePoller.java b/aeron-archiver/src/main/java/io/aeron/archiver/client/ControlResponsePoller.java index 3c04786280..18c37c13ec 100644 --- a/aeron-archiver/src/main/java/io/aeron/archiver/client/ControlResponsePoller.java +++ b/aeron-archiver/src/main/java/io/aeron/archiver/client/ControlResponsePoller.java @@ -192,7 +192,7 @@ private void handleRecordingDescriptor( recordingDescriptorDecoder.mtuLength(), recordingDescriptorDecoder.sessionId(), recordingDescriptorDecoder.streamId(), - recordingDescriptorDecoder.channel(), + recordingDescriptorDecoder.strippedChannel(), recordingDescriptorDecoder.originalChannel(), recordingDescriptorDecoder.sourceIdentity()); } diff --git a/aeron-archiver/src/main/java/io/aeron/archiver/client/RecordingDescriptorConsumer.java b/aeron-archiver/src/main/java/io/aeron/archiver/client/RecordingDescriptorConsumer.java index 73fb4ad27e..d55b612f8d 100644 --- a/aeron-archiver/src/main/java/io/aeron/archiver/client/RecordingDescriptorConsumer.java +++ b/aeron-archiver/src/main/java/io/aeron/archiver/client/RecordingDescriptorConsumer.java @@ -36,7 +36,7 @@ public interface RecordingDescriptorConsumer * @param mtuLength for the recorded publication. * @param sessionId for the recorded publication. * @param streamId for the recorded publication. - * @param channel for the recorded publication. + * @param strippedChannel for the recorded publication. * @param originalChannel for the recorded publication. * @param sourceIdentity for the recorded publication. */ @@ -53,7 +53,7 @@ void onRecordingDescriptor( int mtuLength, int sessionId, int streamId, - String channel, + String strippedChannel, String originalChannel, String sourceIdentity); } diff --git a/aeron-archiver/src/test/java/io/aeron/archiver/ArchiverSystemTest.java b/aeron-archiver/src/test/java/io/aeron/archiver/ArchiverSystemTest.java index 612f7e2560..a82ffc57c3 100644 --- a/aeron-archiver/src/test/java/io/aeron/archiver/ArchiverSystemTest.java +++ b/aeron-archiver/src/test/java/io/aeron/archiver/ArchiverSystemTest.java @@ -465,7 +465,7 @@ public void onRecordingDescriptor( final int mtuLength, final int sessionId, final int streamId, - final String channel, + final String strippedChannel, final String originalChannel, final String sourceIdentity) { diff --git a/aeron-archiver/src/test/java/io/aeron/archiver/CatalogTest.java b/aeron-archiver/src/test/java/io/aeron/archiver/CatalogTest.java index c6b517ea7e..56be5e2c39 100644 --- a/aeron-archiver/src/test/java/io/aeron/archiver/CatalogTest.java +++ b/aeron-archiver/src/test/java/io/aeron/archiver/CatalogTest.java @@ -79,7 +79,7 @@ private void verifyRecordingForId( final long id, final int sessionId, final int streamId, - final String channel, + final String strippedChannel, final String sourceIdentity) throws IOException { @@ -93,8 +93,8 @@ private void verifyRecordingForId( assertEquals(id, recordingDescriptorDecoder.recordingId()); assertEquals(sessionId, recordingDescriptorDecoder.sessionId()); assertEquals(streamId, recordingDescriptorDecoder.streamId()); - assertEquals(channel, recordingDescriptorDecoder.channel()); - assertEquals(channel + "?tag=f", recordingDescriptorDecoder.originalChannel()); + assertEquals(strippedChannel, recordingDescriptorDecoder.strippedChannel()); + assertEquals(strippedChannel + "?tag=f", recordingDescriptorDecoder.originalChannel()); assertEquals(sourceIdentity, recordingDescriptorDecoder.sourceIdentity()); } diff --git a/aeron-archiver/src/test/java/io/aeron/archiver/FailControlResponseListener.java b/aeron-archiver/src/test/java/io/aeron/archiver/FailControlResponseListener.java index 41778aea0d..8d07bffc41 100644 --- a/aeron-archiver/src/test/java/io/aeron/archiver/FailControlResponseListener.java +++ b/aeron-archiver/src/test/java/io/aeron/archiver/FailControlResponseListener.java @@ -50,7 +50,7 @@ public void onRecordingDescriptor( final int mtuLength, final int sessionId, final int streamId, - final String channel, + final String strippedChannel, final String originalChannel, final String sourceIdentity) {