http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterHelper.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterHelper.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterHelper.java deleted file mode 100644 index 0530495..0000000 --- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/cluster/AtomixClusterHelper.java +++ /dev/null @@ -1,83 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 - * - * http://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.apache.camel.component.atomix.cluster; - -import java.io.InputStream; -import java.util.Properties; - -import io.atomix.AtomixReplica; -import io.atomix.catalyst.transport.Address; -import io.atomix.copycat.server.storage.Storage; -import org.apache.camel.CamelContext; -import org.apache.camel.util.ObjectHelper; -import org.apache.camel.util.ResourceHelper; - -public final class AtomixClusterHelper { - private AtomixClusterHelper() { - } - - public static AtomixReplica createReplica(CamelContext camelContext, String address, AtomixClusterConfiguration configuration) throws Exception { - return createReplica(camelContext, new Address(address), configuration); - } - - public static AtomixReplica createReplica(CamelContext camelContext, Address address, AtomixClusterConfiguration configuration) throws Exception { - AtomixReplica atomix = configuration.getAtomix(); - - if (atomix == null) { - final AtomixReplica.Builder atomixBuilder; - - String uri = configuration.getConfigurationUri(); - if (ObjectHelper.isNotEmpty(uri)) { - uri = camelContext.resolvePropertyPlaceholders(uri); - try (InputStream is = ResourceHelper.resolveMandatoryResourceAsInputStream(camelContext, uri)) { - Properties properties = new Properties(); - properties.load(is); - - atomixBuilder = AtomixReplica.builder(address, properties); - } - } else { - atomixBuilder = AtomixReplica.builder(address); - } - - Storage.Builder storageBuilder = Storage.builder(); - ObjectHelper.ifNotEmpty(configuration.getStorageLevel(), storageBuilder::withStorageLevel); - ObjectHelper.ifNotEmpty(configuration.getStoragePath(), storageBuilder::withDirectory); - - atomixBuilder.withStorage(storageBuilder.build()); - - if (configuration.getTransport() != null) { - atomixBuilder.withTransport( - camelContext.getInjector().newInstance(configuration.getTransport()) - ); - } - if (configuration.getClientTransport() != null) { - atomixBuilder.withClientTransport( - camelContext.getInjector().newInstance(configuration.getClientTransport()) - ); - } - if (configuration.getServerTransport() != null) { - atomixBuilder.withServerTransport( - camelContext.getInjector().newInstance(configuration.getServerTransport()) - ); - } - - atomix = atomixBuilder.build(); - } - - return atomix; - } -}
http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterConfiguration.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterConfiguration.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterConfiguration.java new file mode 100644 index 0000000..5cbf30a --- /dev/null +++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterConfiguration.java @@ -0,0 +1,105 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.camel.component.atomix.ha; + +import io.atomix.AtomixReplica; +import io.atomix.catalyst.transport.Transport; +import io.atomix.copycat.server.storage.StorageLevel; +import org.apache.camel.RuntimeCamelException; +import org.apache.camel.component.atomix.AtomixConfiguration; +import org.apache.camel.spi.UriParam; +import org.apache.camel.spi.UriParams; + +@UriParams +public class AtomixClusterConfiguration extends AtomixConfiguration<AtomixReplica> implements Cloneable { + + @UriParam + private Class<? extends Transport> clientTransport; + + @UriParam + private Class<? extends Transport> serverTransport; + + @UriParam + private String storagePath; + + @UriParam(defaultValue = "MEMORY") + private StorageLevel storageLevel = StorageLevel.MEMORY; + + public AtomixClusterConfiguration() { + } + + // ****************************************** + // Properties + // ****************************************** + + + public Class<? extends Transport> getClientTransport() { + return clientTransport; + } + + /** + * The client transport + */ + public void setClientTransport(Class<? extends Transport> clientTransport) { + this.clientTransport = clientTransport; + } + + public Class<? extends Transport> getServerTransport() { + return serverTransport; + } + + /** + * The server transport + */ + public void setServerTransport(Class<? extends Transport> serverTransport) { + this.serverTransport = serverTransport; + } + + public String getStoragePath() { + return storagePath; + } + + /** + * Sets the log directory. + */ + public void setStoragePath(String storagePath) { + this.storagePath = storagePath; + } + + public StorageLevel getStorageLevel() { + return storageLevel; + } + + /** + * Sets the log storage level. + */ + public void setStorageLevel(StorageLevel storageLevel) { + this.storageLevel = storageLevel; + } + + // **************************************** + // Copy + // **************************************** + + public AtomixClusterConfiguration copy() { + try { + return (AtomixClusterConfiguration) super.clone(); + } catch (CloneNotSupportedException e) { + throw new RuntimeCamelException(e); + } + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterHelper.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterHelper.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterHelper.java new file mode 100644 index 0000000..4d41263 --- /dev/null +++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterHelper.java @@ -0,0 +1,83 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.camel.component.atomix.ha; + +import java.io.InputStream; +import java.util.Properties; + +import io.atomix.AtomixReplica; +import io.atomix.catalyst.transport.Address; +import io.atomix.copycat.server.storage.Storage; +import org.apache.camel.CamelContext; +import org.apache.camel.util.ObjectHelper; +import org.apache.camel.util.ResourceHelper; + +public final class AtomixClusterHelper { + private AtomixClusterHelper() { + } + + public static AtomixReplica createReplica(CamelContext camelContext, String address, AtomixClusterConfiguration configuration) throws Exception { + return createReplica(camelContext, new Address(address), configuration); + } + + public static AtomixReplica createReplica(CamelContext camelContext, Address address, AtomixClusterConfiguration configuration) throws Exception { + AtomixReplica atomix = configuration.getAtomix(); + + if (atomix == null) { + final AtomixReplica.Builder atomixBuilder; + + String uri = configuration.getConfigurationUri(); + if (ObjectHelper.isNotEmpty(uri)) { + uri = camelContext.resolvePropertyPlaceholders(uri); + try (InputStream is = ResourceHelper.resolveMandatoryResourceAsInputStream(camelContext, uri)) { + Properties properties = new Properties(); + properties.load(is); + + atomixBuilder = AtomixReplica.builder(address, properties); + } + } else { + atomixBuilder = AtomixReplica.builder(address); + } + + Storage.Builder storageBuilder = Storage.builder(); + ObjectHelper.ifNotEmpty(configuration.getStorageLevel(), storageBuilder::withStorageLevel); + ObjectHelper.ifNotEmpty(configuration.getStoragePath(), storageBuilder::withDirectory); + + atomixBuilder.withStorage(storageBuilder.build()); + + if (configuration.getTransport() != null) { + atomixBuilder.withTransport( + camelContext.getInjector().newInstance(configuration.getTransport()) + ); + } + if (configuration.getClientTransport() != null) { + atomixBuilder.withClientTransport( + camelContext.getInjector().newInstance(configuration.getClientTransport()) + ); + } + if (configuration.getServerTransport() != null) { + atomixBuilder.withServerTransport( + camelContext.getInjector().newInstance(configuration.getServerTransport()) + ); + } + + atomix = atomixBuilder.build(); + } + + return atomix; + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java index bf49eab..06f9432 100644 --- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java +++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterService.java @@ -23,8 +23,6 @@ import io.atomix.catalyst.transport.Address; import io.atomix.catalyst.transport.Transport; import io.atomix.copycat.server.storage.StorageLevel; import org.apache.camel.CamelContext; -import org.apache.camel.component.atomix.cluster.AtomixClusterConfiguration; -import org.apache.camel.component.atomix.cluster.AtomixClusterHelper; import org.apache.camel.impl.ha.AbstractCamelClusterService; import org.apache.camel.util.ObjectHelper; import org.slf4j.Logger; http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java index f7e3be1d..a255446 100644 --- a/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java +++ b/components/camel-atomix/src/main/java/org/apache/camel/component/atomix/ha/AtomixClusterView.java @@ -80,11 +80,20 @@ final class AtomixClusterView extends AbstractCamelClusterView { return new AtomixClusterMember(group, member); } + @SuppressWarnings("unchecked") @Override protected void doStart() throws Exception { if (!localMember.hasJoined()) { LOGGER.debug("Get group {}", getNamespace()); - group = this.atomix.getGroup(getNamespace()).get(); + + final AtomixClusterService service = getClusterService().unwrap(AtomixClusterService.class); + final AtomixClusterConfiguration configuration = service.getConfiguration(); + + group = this.atomix.getGroup( + getNamespace(), + new DistributedGroup.Config(configuration.getResourceConfig(getNamespace())), + new DistributedGroup.Options(configuration.getResourceOptions(getNamespace())) + ).get(); LOGGER.debug("Join group {}", getNamespace()); localMember.join(); http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/TypeConverter ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/TypeConverter b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/TypeConverter new file mode 100644 index 0000000..5992355 --- /dev/null +++ b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/TypeConverter @@ -0,0 +1,18 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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 +# +# http://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. +# + +org.apache.camel.component.atomix.AtomixTypeConverter \ No newline at end of file http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-cluster ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-cluster b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-cluster deleted file mode 100644 index 57df716..0000000 --- a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-cluster +++ /dev/null @@ -1,18 +0,0 @@ -# -# Licensed to the Apache Software Foundation (ASF) under one or more -# contributor license agreements. See the NOTICE file distributed with -# this work for additional information regarding copyright ownership. -# The ASF licenses this file to You 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 -# -# http://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. -# - -class=org.apache.camel.component.atomix.AtomixClusterComponent http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-map ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-map b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-map index 268c354..316fc30 100644 --- a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-map +++ b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-map @@ -15,4 +15,4 @@ # limitations under the License. # -class=org.apache.camel.component.atomix.client.map.AtomixClientMapComponent +class=org.apache.camel.component.atomix.client.map.AtomixMapComponent http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-messaging ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-messaging b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-messaging new file mode 100644 index 0000000..bc99c91 --- /dev/null +++ b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-messaging @@ -0,0 +1,18 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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 +# +# http://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. +# + +class=org.apache.camel.component.atomix.client.messaging.AtomixMessagingComponent http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-multimap ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-multimap b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-multimap new file mode 100644 index 0000000..66ddf18 --- /dev/null +++ b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-multimap @@ -0,0 +1,18 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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 +# +# http://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. +# + +class=org.apache.camel.component.atomix.client.multimap.AtomixMultiMapComponent http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-queue ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-queue b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-queue new file mode 100644 index 0000000..5f3dab1 --- /dev/null +++ b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-queue @@ -0,0 +1,18 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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 +# +# http://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. +# + +class=org.apache.camel.component.atomix.client.queue.AtomixQueueComponent http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-set ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-set b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-set new file mode 100644 index 0000000..c784eca --- /dev/null +++ b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-set @@ -0,0 +1,18 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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 +# +# http://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. +# + +class=org.apache.camel.component.atomix.client.set.AtomixSetComponent http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-value ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-value b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-value new file mode 100644 index 0000000..b24b0fe --- /dev/null +++ b/components/camel-atomix/src/main/resources/META-INF/services/org/apache/camel/component/atomix-value @@ -0,0 +1,18 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You 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 +# +# http://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. +# + +class=org.apache.camel.component.atomix.client.value.AtomixValueComponent http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/AtomixTypeConverterTest.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/AtomixTypeConverterTest.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/AtomixTypeConverterTest.java new file mode 100644 index 0000000..7ec6b48 --- /dev/null +++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/AtomixTypeConverterTest.java @@ -0,0 +1,48 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.camel.component.atomix; + +import io.atomix.catalyst.transport.Address; +import org.apache.camel.TypeConverter; +import org.apache.camel.impl.DefaultCamelContext; +import org.apache.camel.spi.TypeConverterRegistry; +import org.junit.Assert; +import org.junit.Test; + +public class AtomixTypeConverterTest { + + @Test + public void testStringToAddressConversion() throws Exception { + DefaultCamelContext context = null; + + try { + context = new DefaultCamelContext(); + context.start(); + + TypeConverterRegistry registry = context.getTypeConverterRegistry(); + TypeConverter converter = registry.lookup(Address.class, String.class); + + Assert.assertNotNull(converter); + Assert.assertEquals("127.0.0.1", converter.mandatoryConvertTo(Address.class, "127.0.0.1:1234").host()); + Assert.assertEquals(1234, converter.mandatoryConvertTo(Address.class, "127.0.0.1:1234").port()); + } finally { + if (context != null) { + context.stop(); + } + } + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientSpringTestSupport.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientSpringTestSupport.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientSpringTestSupport.java new file mode 100644 index 0000000..450c8a5 --- /dev/null +++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientSpringTestSupport.java @@ -0,0 +1,36 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.camel.component.atomix.client; + +import io.atomix.AtomixClient; +import io.atomix.AtomixReplica; +import io.atomix.catalyst.transport.Address; +import org.apache.camel.test.spring.CamelSpringTestSupport; + +public abstract class AtomixClientSpringTestSupport extends CamelSpringTestSupport { + protected Address getReplicaAddress() { + return applicationContext.getBean(Address.class); + } + + protected AtomixReplica getReplica() { + return applicationContext.getBean(AtomixReplica.class); + } + + protected AtomixClient getClient() { + return applicationContext.getBean(AtomixClient.class); + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientTestSupport.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientTestSupport.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientTestSupport.java index 5251627..f169584 100644 --- a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientTestSupport.java +++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixClientTestSupport.java @@ -16,27 +16,35 @@ */ package org.apache.camel.component.atomix.client; +import java.util.Map; + import io.atomix.AtomixClient; import io.atomix.AtomixReplica; import io.atomix.catalyst.transport.Address; -import org.apache.camel.component.atomix.AtomixHelper; -import org.apache.camel.test.AvailablePortFinder; +import org.apache.camel.Component; +import org.apache.camel.impl.JndiRegistry; import org.apache.camel.test.junit4.CamelTestSupport; public abstract class AtomixClientTestSupport extends CamelTestSupport { - private Address replicaAddress; - private AtomixReplica replica; - private AtomixClient client; + protected Address replicaAddress; + protected AtomixReplica replica; + protected AtomixClient client; @Override - protected void doPreSetup() throws Exception { - replicaAddress = new Address("127.0.0.1", AvailablePortFinder.getNextAvailable()); + protected JndiRegistry createRegistry() throws Exception { + JndiRegistry registry = super.createRegistry(); - replica = AtomixReplica.builder(replicaAddress).withStorage(AtomixHelper.inMemoryStorage()).build(); - replica.bootstrap().join(); + createComponents().entrySet().stream() + .forEach(e -> registry.bind(e.getKey(), e.getValue())); + + return registry; + } - client = AtomixClient.builder().build(); - client.connect(replicaAddress).join(); + @Override + protected void doPreSetup() throws Exception { + replicaAddress = AtomixFactory.address("127.0.0.1"); + replica = AtomixFactory.replica(replicaAddress); + client = AtomixFactory.client(replicaAddress); super.doPreSetup(); } @@ -45,15 +53,20 @@ public abstract class AtomixClientTestSupport extends CamelTestSupport { public void tearDown() throws Exception { if (client != null) { client.close().join(); + client = null; } + if (replica != null) { replica.shutdown().join(); replica.leave().join(); + replica = null; } super.tearDown(); } + protected abstract Map<String, Component> createComponents(); + // ************************************* // properties // ************************************* http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixFactory.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixFactory.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixFactory.java new file mode 100644 index 0000000..13862a7 --- /dev/null +++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/AtomixFactory.java @@ -0,0 +1,47 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.camel.component.atomix.client; + +import io.atomix.AtomixClient; +import io.atomix.AtomixReplica; +import io.atomix.catalyst.transport.Address; +import org.apache.camel.component.atomix.AtomixHelper; +import org.apache.camel.test.AvailablePortFinder; + +public final class AtomixFactory { + + private AtomixFactory() { + } + + public static Address address(String host) { + return new Address(host, AvailablePortFinder.getNextAvailable()); + } + + public static AtomixReplica replica(Address address) { + AtomixReplica replica = AtomixReplica.builder(address).withStorage(AtomixHelper.inMemoryStorage()).build(); + replica.bootstrap().join(); + + return replica; + } + + public static AtomixClient client(Address address) { + AtomixClient client = AtomixClient.builder().build(); + client.connect(address).join(); + + return client; + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConsumerTest.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConsumerTest.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConsumerTest.java deleted file mode 100644 index 8a7728a..0000000 --- a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixClientMapConsumerTest.java +++ /dev/null @@ -1,115 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 - * - * http://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.apache.camel.component.atomix.client.map; - -import java.util.Collections; -import java.util.UUID; - -import io.atomix.collections.DistributedMap; -import org.apache.camel.RoutesBuilder; -import org.apache.camel.builder.RouteBuilder; -import org.apache.camel.component.atomix.client.AtomixClientConstants; -import org.apache.camel.component.atomix.client.AtomixClientTestSupport; -import org.apache.camel.component.mock.MockEndpoint; -import org.apache.camel.impl.JndiRegistry; -import org.junit.Ignore; -import org.junit.Test; - -@Ignore -public class AtomixClientMapConsumerTest extends AtomixClientTestSupport { - private static final String MAP_NAME = UUID.randomUUID().toString(); - private DistributedMap<Object, Object> map; - - // ************************************ - // Setup - // ************************************ - - @Override - protected JndiRegistry createRegistry() throws Exception { - JndiRegistry registry = super.createRegistry(); - - AtomixClientMapComponent component = new AtomixClientMapComponent(); - component.setNodes(Collections.singletonList(getReplicaAddress())); - - registry.bind("atomix-map", component); - - return registry; - } - - @Override - protected void doPreSetup() throws Exception { - super.doPreSetup(); - - map = getClient().getMap(MAP_NAME).join(); - } - - @Override - public void tearDown() throws Exception { - map.close(); - - super.tearDown(); - } - - // ************************************ - // Test - // ************************************ - - @Test - public void test() throws Exception { - String key = context().getUuidGenerator().generateUuid(); - String put = context().getUuidGenerator().generateUuid(); - String upd = context().getUuidGenerator().generateUuid(); - - MockEndpoint mock = getMockEndpoint("mock:result"); - mock.assertExchangeReceived(3); - mock.expectedBodiesReceived(put, upd, upd); - mock.expectedHeaderReceived(AtomixClientConstants.RESOURCE_KEY, key); - - map.put(key, put).join(); - map.replace(key, upd).join(); - map.remove(key).join(); - - mock.assertIsSatisfied(); - - assertEquals( - DistributedMap.Events.ADD, - mock.getExchanges().get(0).getIn().getHeader(AtomixClientConstants.EVENT_TYPE) - ); - assertEquals( - DistributedMap.Events.UPDATE, - mock.getExchanges().get(1).getIn().getHeader(AtomixClientConstants.EVENT_TYPE) - ); - assertEquals( - DistributedMap.Events.REMOVE, - mock.getExchanges().get(2).getIn().getHeader(AtomixClientConstants.EVENT_TYPE) - ); - } - - // ************************************ - // Routes - // ************************************ - - @Override - protected RoutesBuilder createRouteBuilder() throws Exception { - return new RouteBuilder() { - public void configure() { - fromF("atomix-map:%s", MAP_NAME) - .to("mock:result"); - } - }; - } -} http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixClientMapProducerTest.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixClientMapProducerTest.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixClientMapProducerTest.java deleted file mode 100644 index f46d628..0000000 --- a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixClientMapProducerTest.java +++ /dev/null @@ -1,398 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You 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 - * - * http://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.apache.camel.component.atomix.client.map; - -import java.time.Duration; -import java.util.Collection; -import java.util.Collections; -import java.util.Set; -import java.util.UUID; -import java.util.concurrent.CountDownLatch; - -import io.atomix.collections.DistributedMap; -import org.apache.camel.EndpointInject; -import org.apache.camel.FluentProducerTemplate; -import org.apache.camel.Message; -import org.apache.camel.RoutesBuilder; -import org.apache.camel.builder.RouteBuilder; -import org.apache.camel.component.atomix.client.AtomixClientAction; -import org.apache.camel.component.atomix.client.AtomixClientConstants; -import org.apache.camel.component.atomix.client.AtomixClientTestSupport; -import org.apache.camel.impl.JndiRegistry; -import org.junit.Test; - -import static org.hamcrest.core.Is.is; - -public class AtomixClientMapProducerTest extends AtomixClientTestSupport { - private static final String MAP_NAME = UUID.randomUUID().toString(); - private DistributedMap<Object, Object> map; - - @EndpointInject(uri = "direct:start") - private FluentProducerTemplate fluent; - - // ************************************ - // Setup - // ************************************ - - @Override - protected JndiRegistry createRegistry() throws Exception { - JndiRegistry registry = super.createRegistry(); - - AtomixClientMapComponent component = new AtomixClientMapComponent(); - component.setNodes(Collections.singletonList(getReplicaAddress())); - - registry.bind("atomix-map", component); - - return registry; - } - - @Override - protected void doPreSetup() throws Exception { - super.doPreSetup(); - - map = getClient().getMap(MAP_NAME).join(); - } - - @Override - public void tearDown() throws Exception { - map.close(); - - super.tearDown(); - } - - // ************************************ - // Test - // ************************************ - - @Test - public void testPut() throws Exception { - final String key = context().getUuidGenerator().generateUuid(); - final String val = context().getUuidGenerator().generateUuid(); - - Message result; - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.PUT) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withHeader(AtomixClientConstants.RESOURCE_TTL, Duration.ofMillis(250)) - .withBody(val) - .request(Message.class); - - assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(val, result.getBody()); - assertEquals(val, map.get(key).join()); - - CountDownLatch latch = new CountDownLatch(1); - map.onRemove(e -> latch.countDown()); - - latch.await(); - - assertFalse(val, map.containsKey(key).join()); - } - - @Test - public void testPutIfAbsent() throws Exception { - final String key = context().getUuidGenerator().generateUuid(); - final String val1 = context().getUuidGenerator().generateUuid(); - final String val2 = context().getUuidGenerator().generateUuid(); - - Message result; - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.PUT_IF_ABSENT) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withBody(val1) - .request(Message.class); - - assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(val1, result.getBody()); - assertEquals(val1, map.get(key).join()); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.PUT_IF_ABSENT) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withBody(val2) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(val1, result.getBody()); - assertEquals(val1, map.get(key).join()); - } - - @Test - public void testGet() throws Exception { - final String key = context().getUuidGenerator().generateUuid(); - final String val = context().getUuidGenerator().generateUuid(); - - Message result; - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.GET) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .request(Message.class); - - assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertFalse(map.containsKey(key).join()); - - map.put(key, val).join(); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.GET) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(val, result.getBody(String.class)); - assertTrue(map.containsKey(key).join()); - } - - @Test - public void testSizeClearIsEmpty() throws Exception { - final String key = context().getUuidGenerator().generateUuid(); - final String val = context().getUuidGenerator().generateUuid(); - - Message result; - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.SIZE) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(0, result.getBody(Integer.class).intValue()); - assertEquals(map.size().join(), result.getBody(Integer.class)); - - map.put(key, val).join(); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.SIZE) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(1, result.getBody(Integer.class).intValue()); - assertEquals(map.size().join(), result.getBody(Integer.class)); - - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.IS_EMPTY) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertFalse(result.getBody(Boolean.class)); - assertFalse(map.isEmpty().join()); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.CLEAR) - .request(Message.class); - - assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(0, map.size().join().intValue()); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.IS_EMPTY) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertTrue(result.getBody(Boolean.class)); - assertTrue(map.isEmpty().join()); - } - - @Test - public void testContainsKey() throws Exception { - final String key = context().getUuidGenerator().generateUuid(); - final String val = context().getUuidGenerator().generateUuid(); - - Message result; - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.CONTAINS_KEY) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withBody(val) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertFalse(result.getBody(Boolean.class)); - assertFalse(map.containsKey(key).join()); - - map.put(key, val).join(); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.CONTAINS_KEY) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withBody(val) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertTrue(result.getBody(Boolean.class)); - assertTrue(map.containsKey(key).join()); - } - - @Test - public void testContainsValue() throws Exception { - final String key = context().getUuidGenerator().generateUuid(); - final String val = context().getUuidGenerator().generateUuid(); - - Message result; - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.CONTAINS_VALUE) - .withHeader(AtomixClientConstants.RESOURCE_VALUE, val) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertFalse(result.getBody(Boolean.class)); - assertFalse(map.containsKey(key).join()); - - map.put(key, val).join(); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.CONTAINS_VALUE) - .withHeader(AtomixClientConstants.RESOURCE_VALUE, val) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertTrue(result.getBody(Boolean.class)); - assertTrue(map.containsValue(val).join()); - } - - @Test - public void testRemove() throws Exception { - final String key = context().getUuidGenerator().generateUuid(); - final String val = context().getUuidGenerator().generateUuid(); - - map.put(key, val).join(); - - Message result; - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.REMOVE) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withHeader(AtomixClientConstants.RESOURCE_VALUE, context().getUuidGenerator().generateUuid()) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertFalse(result.getBody(Boolean.class)); - assertTrue(map.containsKey(key).join()); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.REMOVE) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withHeader(AtomixClientConstants.RESOURCE_VALUE, val) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertTrue(result.getBody(Boolean.class)); - assertFalse(map.containsKey(key).join()); - - map.put(key, val).join(); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.REMOVE) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(val, result.getBody(String.class)); - assertFalse(map.containsKey(key).join()); - } - - @Test - public void testReplace() throws Exception { - final String key = context().getUuidGenerator().generateUuid(); - final String oldVal = context().getUuidGenerator().generateUuid(); - final String newVal = context().getUuidGenerator().generateUuid(); - - map.put(key, oldVal).join(); - - Message result; - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.REPLACE) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withHeader(AtomixClientConstants.RESOURCE_OLD_VALUE, context().getUuidGenerator().generateUuid()) - .withBody(newVal) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertFalse(result.getBody(Boolean.class)); - assertEquals(oldVal, map.get(key).join()); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.REPLACE) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withHeader(AtomixClientConstants.RESOURCE_OLD_VALUE, oldVal) - .withBody(newVal) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertTrue(result.getBody(Boolean.class)); - assertEquals(newVal, map.get(key).join()); - - map.put(key, oldVal).join(); - - result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.REPLACE) - .withHeader(AtomixClientConstants.RESOURCE_KEY, key) - .withBody(newVal) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(oldVal, result.getBody(String.class)); - assertEquals(newVal, map.get(key).join()); - } - - @Test - public void testValues() throws Exception { - map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); - map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); - map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); - - Message result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.VALUES) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertThat(map.values().join(), is(result.getBody(Collection.class))); - } - - @Test - public void testEntrySet() throws Exception { - map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); - map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); - map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); - - Message result = fluent.clearAll() - .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixClientAction.ENTRY_SET) - .request(Message.class); - - assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); - assertEquals(map.entrySet().join().size(), result.getBody(Set.class).size()); - } - - // ************************************ - // Routes - // ************************************ - - @Override - protected RoutesBuilder createRouteBuilder() throws Exception { - return new RouteBuilder() { - public void configure() { - from("direct:start") - .toF("atomix-map:%s", MAP_NAME); - } - }; - } -} http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapConsumerTest.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapConsumerTest.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapConsumerTest.java new file mode 100644 index 0000000..14faadc --- /dev/null +++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapConsumerTest.java @@ -0,0 +1,136 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.camel.component.atomix.client.map; + +import java.util.Collections; +import java.util.Map; +import java.util.UUID; + +import io.atomix.collections.DistributedMap; +import org.apache.camel.Component; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.atomix.client.AtomixClientConstants; +import org.apache.camel.component.atomix.client.AtomixClientTestSupport; +import org.apache.camel.component.mock.MockEndpoint; +import org.junit.Test; + +public class AtomixMapConsumerTest extends AtomixClientTestSupport { + private static final String MAP_NAME = UUID.randomUUID().toString(); + private static final String KEY_NAME = UUID.randomUUID().toString(); + private DistributedMap<Object, Object> map; + + // ************************************ + // Setup + // ************************************ + + @Override + protected Map<String, Component> createComponents() { + AtomixMapComponent component = new AtomixMapComponent(); + component.setNodes(Collections.singletonList(getReplicaAddress())); + + return Collections.singletonMap("atomix-map", component); + } + + @Override + protected void doPreSetup() throws Exception { + super.doPreSetup(); + + map = getClient().getMap(MAP_NAME).join(); + } + + @Override + public void tearDown() throws Exception { + map.close(); + + super.tearDown(); + } + + // ************************************ + // Test + // ************************************ + + @Test + public void testEvents() throws Exception { + String key = context().getUuidGenerator().generateUuid(); + String put = context().getUuidGenerator().generateUuid(); + String upd = context().getUuidGenerator().generateUuid(); + + MockEndpoint mock1 = getMockEndpoint("mock:result"); + mock1.expectedMessageCount(6); + MockEndpoint mock2 = getMockEndpoint("mock:result-key"); + mock2.expectedMessageCount(2); + + mock1.message(0).body().isEqualTo(put); + mock1.message(0).header(AtomixClientConstants.EVENT_TYPE).isEqualTo(DistributedMap.Events.ADD); + mock1.message(0).header(AtomixClientConstants.RESOURCE_KEY).isEqualTo(key); + + mock1.message(1).body().isEqualTo(put); + mock1.message(1).header(AtomixClientConstants.EVENT_TYPE).isEqualTo(DistributedMap.Events.UPDATE); + mock1.message(1).header(AtomixClientConstants.RESOURCE_KEY).isEqualTo(key); + + mock1.message(2).body().isEqualTo(upd); + mock1.message(2).header(AtomixClientConstants.EVENT_TYPE).isEqualTo(DistributedMap.Events.UPDATE); + mock1.message(2).header(AtomixClientConstants.RESOURCE_KEY).isEqualTo(key); + + mock1.message(3).body().isEqualTo(upd); + mock1.message(3).header(AtomixClientConstants.EVENT_TYPE).isEqualTo(DistributedMap.Events.REMOVE); + mock1.message(3).header(AtomixClientConstants.RESOURCE_KEY).isEqualTo(key); + + mock1.message(4).body().isEqualTo(put); + mock1.message(4).header(AtomixClientConstants.EVENT_TYPE).isEqualTo(DistributedMap.Events.ADD); + mock1.message(4).header(AtomixClientConstants.RESOURCE_KEY).isEqualTo(KEY_NAME); + + mock1.message(5).body().isEqualTo(put); + mock1.message(5).header(AtomixClientConstants.EVENT_TYPE).isEqualTo(DistributedMap.Events.REMOVE); + mock1.message(5).header(AtomixClientConstants.RESOURCE_KEY).isEqualTo(KEY_NAME); + + mock2.message(0).body().isEqualTo(put); + mock2.message(0).header(AtomixClientConstants.EVENT_TYPE).isEqualTo(DistributedMap.Events.ADD); + mock2.message(0).header(AtomixClientConstants.RESOURCE_KEY).isEqualTo(KEY_NAME); + + mock2.message(1).body().isEqualTo(put); + mock2.message(1).header(AtomixClientConstants.EVENT_TYPE).isEqualTo(DistributedMap.Events.REMOVE); + mock2.message(1).header(AtomixClientConstants.RESOURCE_KEY).isEqualTo(KEY_NAME); + + map.put(key, put).join(); + map.put(key, put).join(); + map.replace(key, upd).join(); + map.remove(key).join(); + map.put(KEY_NAME, put).join(); + map.remove(KEY_NAME).join(); + + mock1.assertIsSatisfied(); + mock2.assertIsSatisfied(); + } + + // ************************************ + // Routes + // ************************************ + + @Override + protected RoutesBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + public void configure() { + fromF("atomix-map:%s", MAP_NAME) + .to("mock:result"); + fromF("atomix-map:%s?key=%s", MAP_NAME, KEY_NAME) + .to("mock:result-key"); + } + }; + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapNodesProducerTest.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapNodesProducerTest.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapNodesProducerTest.java new file mode 100644 index 0000000..85d1523 --- /dev/null +++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapNodesProducerTest.java @@ -0,0 +1,100 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.camel.component.atomix.client.map; + +import java.util.Collections; +import java.util.Map; +import java.util.UUID; + +import io.atomix.collections.DistributedMap; +import org.apache.camel.Component; +import org.apache.camel.EndpointInject; +import org.apache.camel.FluentProducerTemplate; +import org.apache.camel.Message; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.atomix.client.AtomixClientConstants; +import org.apache.camel.component.atomix.client.AtomixClientTestSupport; +import org.junit.Test; + +public class AtomixMapNodesProducerTest extends AtomixClientTestSupport { + + private static final String MAP_NAME = UUID.randomUUID().toString(); + private DistributedMap<Object, Object> map; + + @EndpointInject(uri = "direct:start") + private FluentProducerTemplate fluent; + + // ************************************ + // Setup + // ************************************ + + @Override + protected Map<String, Component> createComponents() { + return Collections.emptyMap(); + } + + @Override + protected void doPreSetup() throws Exception { + super.doPreSetup(); + + map = getClient().getMap(MAP_NAME).join(); + } + + @Override + public void tearDown() throws Exception { + map.close(); + + super.tearDown(); + } + + // ************************************ + // Test + // ************************************ + + @Test + public void testPut() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String val = context().getUuidGenerator().generateUuid(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.PUT) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withBody(val) + .request(Message.class); + + assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(val, result.getBody()); + assertEquals(val, map.get(key).join()); + } + + // ************************************ + // Routes + // ************************************ + + @Override + protected RoutesBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + public void configure() { + from("direct:start") + .toF("atomix-map:%s?nodes=%s:%d", MAP_NAME, replicaAddress.host(), replicaAddress.port()); + } + }; + } +} http://git-wip-us.apache.org/repos/asf/camel/blob/3756fba2/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapProducerTest.java ---------------------------------------------------------------------- diff --git a/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapProducerTest.java b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapProducerTest.java new file mode 100644 index 0000000..bf93cf8 --- /dev/null +++ b/components/camel-atomix/src/test/java/org/apache/camel/component/atomix/client/map/AtomixMapProducerTest.java @@ -0,0 +1,425 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 + * + * http://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.apache.camel.component.atomix.client.map; + +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; + +import io.atomix.collections.DistributedMap; +import org.apache.camel.Component; +import org.apache.camel.EndpointInject; +import org.apache.camel.FluentProducerTemplate; +import org.apache.camel.Message; +import org.apache.camel.RoutesBuilder; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.atomix.client.AtomixClientConstants; +import org.apache.camel.component.atomix.client.AtomixClientTestSupport; +import org.junit.Test; + +import static org.hamcrest.core.Is.is; + +public class AtomixMapProducerTest extends AtomixClientTestSupport { + private static final String MAP_NAME = UUID.randomUUID().toString(); + private DistributedMap<Object, Object> map; + + @EndpointInject(uri = "direct:start") + private FluentProducerTemplate fluent; + + // ************************************ + // Setup + // ************************************ + + @Override + protected Map<String, Component> createComponents() { + AtomixMapComponent component = new AtomixMapComponent(); + component.setNodes(Collections.singletonList(getReplicaAddress())); + + return Collections.singletonMap("atomix-map", component); + } + + @Override + protected void doPreSetup() throws Exception { + super.doPreSetup(); + + map = getClient().getMap(MAP_NAME).join(); + } + + @Override + public void tearDown() throws Exception { + map.close(); + + super.tearDown(); + } + + // ************************************ + // Test + // ************************************ + + @Test + public void testPut() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String val = context().getUuidGenerator().generateUuid(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.PUT) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withBody(val) + .request(Message.class); + + assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(val, result.getBody()); + assertEquals(val, map.get(key).join()); + } + + @Test + public void testPutWithTTL() throws Exception { + final String key1 = context().getUuidGenerator().generateUuid(); + final String key2 = context().getUuidGenerator().generateUuid(); + final String val = context().getUuidGenerator().generateUuid(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.PUT) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key1) + .withHeader(AtomixClientConstants.RESOURCE_TTL, "1s") + .withBody(val) + .request(Message.class); + + assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(val, result.getBody()); + assertEquals(val, map.get(key1).join()); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.PUT) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key2) + .withHeader(AtomixClientConstants.RESOURCE_TTL, "250") + .withBody(val) + .request(Message.class); + + assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(val, result.getBody()); + assertEquals(val, map.get(key2).join()); + + CountDownLatch latch = new CountDownLatch(2); + map.onRemove(key1, e -> latch.countDown()); + map.onRemove(key2, e -> latch.countDown()); + + latch.await(); + + assertFalse(map.containsKey(key1).join()); + assertFalse(map.containsKey(key2).join()); + } + + @Test + public void testPutIfAbsent() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String val1 = context().getUuidGenerator().generateUuid(); + final String val2 = context().getUuidGenerator().generateUuid(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.PUT_IF_ABSENT) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withBody(val1) + .request(Message.class); + + assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(val1, result.getBody()); + assertEquals(val1, map.get(key).join()); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.PUT_IF_ABSENT) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withBody(val2) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(val1, result.getBody()); + assertEquals(val1, map.get(key).join()); + } + + @Test + public void testGet() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String val = context().getUuidGenerator().generateUuid(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.GET) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .request(Message.class); + + assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertFalse(map.containsKey(key).join()); + + map.put(key, val).join(); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.GET) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(val, result.getBody(String.class)); + assertTrue(map.containsKey(key).join()); + } + + @Test + public void testSizeClearIsEmpty() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String val = context().getUuidGenerator().generateUuid(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.SIZE) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(0, result.getBody(Integer.class).intValue()); + assertEquals(map.size().join(), result.getBody(Integer.class)); + + map.put(key, val).join(); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.SIZE) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(1, result.getBody(Integer.class).intValue()); + assertEquals(map.size().join(), result.getBody(Integer.class)); + + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.IS_EMPTY) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertFalse(result.getBody(Boolean.class)); + assertFalse(map.isEmpty().join()); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.CLEAR) + .request(Message.class); + + assertFalse(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(0, map.size().join().intValue()); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.IS_EMPTY) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertTrue(result.getBody(Boolean.class)); + assertTrue(map.isEmpty().join()); + } + + @Test + public void testContainsKey() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String val = context().getUuidGenerator().generateUuid(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.CONTAINS_KEY) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withBody(val) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertFalse(result.getBody(Boolean.class)); + assertFalse(map.containsKey(key).join()); + + map.put(key, val).join(); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.CONTAINS_KEY) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withBody(val) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertTrue(result.getBody(Boolean.class)); + assertTrue(map.containsKey(key).join()); + } + + @Test + public void testContainsValue() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String val = context().getUuidGenerator().generateUuid(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.CONTAINS_VALUE) + .withHeader(AtomixClientConstants.RESOURCE_VALUE, val) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertFalse(result.getBody(Boolean.class)); + assertFalse(map.containsKey(key).join()); + + map.put(key, val).join(); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.CONTAINS_VALUE) + .withHeader(AtomixClientConstants.RESOURCE_VALUE, val) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertTrue(result.getBody(Boolean.class)); + assertTrue(map.containsValue(val).join()); + } + + @Test + public void testRemove() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String val = context().getUuidGenerator().generateUuid(); + + map.put(key, val).join(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.REMOVE) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withHeader(AtomixClientConstants.RESOURCE_VALUE, context().getUuidGenerator().generateUuid()) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertFalse(result.getBody(Boolean.class)); + assertTrue(map.containsKey(key).join()); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.REMOVE) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withHeader(AtomixClientConstants.RESOURCE_VALUE, val) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertTrue(result.getBody(Boolean.class)); + assertFalse(map.containsKey(key).join()); + + map.put(key, val).join(); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.REMOVE) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(val, result.getBody(String.class)); + assertFalse(map.containsKey(key).join()); + } + + @Test + public void testReplace() throws Exception { + final String key = context().getUuidGenerator().generateUuid(); + final String oldVal = context().getUuidGenerator().generateUuid(); + final String newVal = context().getUuidGenerator().generateUuid(); + + map.put(key, oldVal).join(); + + Message result; + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.REPLACE) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withHeader(AtomixClientConstants.RESOURCE_OLD_VALUE, context().getUuidGenerator().generateUuid()) + .withBody(newVal) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertFalse(result.getBody(Boolean.class)); + assertEquals(oldVal, map.get(key).join()); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.REPLACE) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withHeader(AtomixClientConstants.RESOURCE_OLD_VALUE, oldVal) + .withBody(newVal) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertTrue(result.getBody(Boolean.class)); + assertEquals(newVal, map.get(key).join()); + + map.put(key, oldVal).join(); + + result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.REPLACE) + .withHeader(AtomixClientConstants.RESOURCE_KEY, key) + .withBody(newVal) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(oldVal, result.getBody(String.class)); + assertEquals(newVal, map.get(key).join()); + } + + @Test + public void testValues() throws Exception { + map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); + map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); + map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); + + Message result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.VALUES) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertThat(map.values().join(), is(result.getBody(Collection.class))); + } + + @Test + public void testEntrySet() throws Exception { + map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); + map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); + map.put(context().getUuidGenerator().generateUuid(), context().getUuidGenerator().generateUuid()).join(); + + Message result = fluent.clearAll() + .withHeader(AtomixClientConstants.RESOURCE_ACTION, AtomixMap.Action.ENTRY_SET) + .request(Message.class); + + assertTrue(result.getHeader(AtomixClientConstants.RESOURCE_ACTION_HAS_RESULT, Boolean.class)); + assertEquals(map.entrySet().join().size(), result.getBody(Set.class).size()); + } + + // ************************************ + // Routes + // ************************************ + + @Override + protected RoutesBuilder createRouteBuilder() throws Exception { + return new RouteBuilder() { + public void configure() { + from("direct:start") + .toF("atomix-map:%s", MAP_NAME); + } + }; + } +}
