diff --git a/src/main/java/io/github/eocqrs/kafka/Consumer.java b/src/main/java/io/github/eocqrs/kafka/Consumer.java index 37da4dad..d08ef69b 100644 --- a/src/main/java/io/github/eocqrs/kafka/Consumer.java +++ b/src/main/java/io/github/eocqrs/kafka/Consumer.java @@ -73,6 +73,7 @@ public interface Consumer extends Closeable { /** * Unsubscribe. + * @throws Exception When something went wrong. */ - void unsubscribe(); + void unsubscribe() throws Exception; } diff --git a/src/main/java/io/github/eocqrs/kafka/fake/FkConsumer.java b/src/main/java/io/github/eocqrs/kafka/fake/FkConsumer.java index 44dd36b0..9c58e451 100644 --- a/src/main/java/io/github/eocqrs/kafka/fake/FkConsumer.java +++ b/src/main/java/io/github/eocqrs/kafka/fake/FkConsumer.java @@ -109,12 +109,18 @@ public ConsumerRecords records( throw new UnsupportedOperationException("#records()"); } - /* - * @todo #54:60m/DEV Fake unsubscribe is not implemented - */ @Override - public void unsubscribe() { - throw new UnsupportedOperationException("#unsubscribe()"); + public void unsubscribe() throws Exception { + while ( + !this.broker.data( + "broker/subs/sub[consumer = '%s']/consumer/text()" + .formatted( + this.id + ) + ).isEmpty() + ) { + this.broker.with(new UnsubscribeDirs(this.id).value()); + } } @Override diff --git a/src/main/java/io/github/eocqrs/kafka/fake/UnsubscribeDirs.java b/src/main/java/io/github/eocqrs/kafka/fake/UnsubscribeDirs.java new file mode 100644 index 00000000..5102e317 --- /dev/null +++ b/src/main/java/io/github/eocqrs/kafka/fake/UnsubscribeDirs.java @@ -0,0 +1,63 @@ +/* + * Copyright (c) 2023 Aliaksei Bialiauski, EO-CQRS + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package io.github.eocqrs.kafka.fake; + +import org.cactoos.Scalar; +import org.xembly.Directives; + +import java.util.UUID; + +/** + * Unsubscribe Directives. + * + * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com) + * @since 0.3.5 + */ +public final class UnsubscribeDirs implements Scalar { + + /** + * Consumer ID. + */ + private final UUID consumer; + + /** + * Ctor. + * + * @param cnsmr Consumer ID + */ + public UnsubscribeDirs(final UUID cnsmr) { + this.consumer = cnsmr; + } + + @Override + public Directives value() throws Exception { + return new Directives() + .xpath("broker/subs/sub[consumer = '%s']/consumer" + .formatted( + this.consumer + ) + ) + .remove() + .remove(); + } +} diff --git a/src/test/java/io/github/eocqrs/kafka/fake/FkConsumerTest.java b/src/test/java/io/github/eocqrs/kafka/fake/FkConsumerTest.java index a21a7417..7f7853a2 100644 --- a/src/test/java/io/github/eocqrs/kafka/fake/FkConsumerTest.java +++ b/src/test/java/io/github/eocqrs/kafka/fake/FkConsumerTest.java @@ -43,6 +43,7 @@ import java.io.IOException; import java.time.Duration; import java.util.Collection; +import java.util.Map; import java.util.UUID; import java.util.logging.Level; @@ -241,6 +242,87 @@ public void onPartitionsAssigned(final Collection collection) { ); } + @Test + void unsubscribes() throws Exception { + final String topic = "unsubscribe.test"; + final UUID uuid = UUID.randomUUID(); + final Consumer consumer = + new FkConsumer<>( + uuid, + this.broker + ); + consumer.subscribe(topic); + consumer.unsubscribe(); + MatcherAssert.assertThat( + "Consumer ID in subscription is not present", + this.broker.data( + "broker/subs/sub[consumer = '%s']/consumer/text()" + .formatted( + uuid + ) + ) + .isEmpty(), + Matchers.equalTo(true) + ); + MatcherAssert.assertThat( + "Topic in subscription is not present", + this.broker.data( + "broker/subs/sub[topic = '%s']/topic/text()" + .formatted( + topic + ) + ) + .isEmpty(), + Matchers.equalTo(true) + ); + consumer.close(); + } + + @Test + void unsubscribesWithSecondConsumerExisting() throws Exception { + final String topic = "unsubscribes.with.second.consumer.existing"; + final UUID firstID = UUID.fromString("f3000fb7-b9fb-42d0-8210-f09a58c44a1f"); + final UUID secondID = UUID.fromString("69a4cd5a-afdb-456c-9ade-658569f52d7b"); + final Consumer first = + new FkConsumer<>( + firstID, + this.broker + ); + final Consumer second = + new FkConsumer<>( + secondID, + this.broker + ); + first.subscribe(topic); + second.subscribe(topic); + first.unsubscribe(); + MatcherAssert.assertThat( + "No such subscription with first Consumer ID", + this.broker.data( + "broker/subs/sub[topic = '%s' and consumer = '%s']/topic/text()" + .formatted( + topic, + firstID + ) + ).isEmpty(), + Matchers.equalTo(true) + ); + MatcherAssert.assertThat( + "Topic with subscription exists with second Consumer ID", + this.broker.data( + "broker/subs/sub[topic = '%s' and consumer = '%s']/topic/text()" + .formatted( + topic, + secondID + ) + ) + .isEmpty(), + Matchers.equalTo(false) + ); + first.close(); + second.close(); + } + @Test void createsFakeConsumer() { final FkConsumer consumer = @@ -254,10 +336,6 @@ void createsFakeConsumer() { UnsupportedOperationException.class, () -> consumer.records("123", Duration.ofMillis(100L)) ); - assertThrows( - UnsupportedOperationException.class, - consumer::unsubscribe - ); } @Test diff --git a/src/test/java/io/github/eocqrs/kafka/fake/UnsubscribeDirsTest.java b/src/test/java/io/github/eocqrs/kafka/fake/UnsubscribeDirsTest.java new file mode 100644 index 00000000..d78aab97 --- /dev/null +++ b/src/test/java/io/github/eocqrs/kafka/fake/UnsubscribeDirsTest.java @@ -0,0 +1,53 @@ +/* + * Copyright (c) 2023 Aliaksei Bialiauski, EO-CQRS + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package io.github.eocqrs.kafka.fake; + +import org.hamcrest.MatcherAssert; +import org.hamcrest.Matchers; +import org.junit.jupiter.api.Test; + +import java.util.UUID; + +/** + * Test case for {@link UnsubscribeDirs}. + * + * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com) + * @since 0.3.5 + */ +final class UnsubscribeDirsTest { + + @Test + void dirsInRightFormat() throws Exception { + final UUID uuid = UUID.fromString("1ce12119-7ccc-46fc-a993-36d5b815a7b5"); + final String directives = "XPATH \"broker/subs/sub" + + "[consumer = '1ce12119-7ccc-46fc-a993-36d5b815a7b5']/consumer\";" + + "\n1:REMOVE;REMOVE;"; + MatcherAssert.assertThat( + "Directives in right format", + new UnsubscribeDirs(uuid) + .value() + .toString(), + Matchers.equalTo(directives) + ); + } +}