Distributed Maps

public class DistributedMapExample {
    /**
     * This example demonstrates the usage of the IgniteStructures API.
     */

    public static void main(String[] args) throws Exception {

        try (IgniteClient client1 = IgniteClient.builder()
                .addresses("127.0.0.1:10800")
                .build();
                IgniteClient client2 = IgniteClient.builder()
                        .addresses("127.0.0.1:10801")
                        .build()) {

            //--------------------------------------------------------------------------------------
            //
            // Base put and get operations with distributed maps.
            //
            //--------------------------------------------------------------------------------------

        CompletableFuture<IgniteMap<Integer, String>> mapFuture = client1.structures().getOrCreateMapAsync(
                MapConfiguration.builder("map1", Integer.class, String.class)
                        .distributionZone("ZONENAME")
                        .storageProfile("profileName")
                        .build());

        IgniteMap<Integer, String> mapFromNode1 = mapFuture.get(10, TimeUnit.SECONDS);

        basicOperationsSync(mapFromNode1);

        basicOperationsAsync(mapFromNode1);

        mapFromNode1.putAsync(1, "val1").thenCompose(prev -> {
            if (prev != null) {
                throw new RuntimeException("Previous value should be null");
            }

            return mapFromNode1.getAsync(1)
                    .thenAccept(newValue -> System.out.println("New value from node 1 after putAsync: " + newValue));
        }).get(10, TimeUnit.SECONDS);

        IgniteMap<Integer, String> mapFromNode2 = client2.structures().getOrCreateMap(MapConfiguration.builder("map1", Integer.class, String.class)
                .distributionZone("ZONENAME")
                .storageProfile("profileName")
                .build());

        mapFromNode2.getAsync(1).thenAccept(v -> System.out.println("Before putAsync - node2 get: " + v)).get(10, TimeUnit.SECONDS);

        mapFromNode2.putAsync(1, "val1").thenCompose(prev -> {
            System.out.println("Result of putAsync(previous): " + prev);

            return mapFromNode2.getAsync(1).thenAccept(newValue ->
                    System.out.println("After putAsync - node2 get: " + newValue));
        }).get(10, TimeUnit.SECONDS);


        //--------------------------------------------------------------------------------------
        //
        // Base put and get operations with custom key/value classes based map.
        //
        //--------------------------------------------------------------------------------------
        CompletableFuture<IgniteMap<KeyClass, ValueClass>> mapFuture2 = client1.structures()
                .getOrCreateMapAsync("map2", new KeySerializer(), new ValueSerializer());

        IgniteMap<KeyClass, ValueClass> mapWithCustomSerialization = mapFuture2.get(10, TimeUnit.SECONDS);

        KeyClass key1 = new KeyClass("keyStr1", 1);
        ValueClass value1 = new ValueClass(List.of("valStr1"));
        CompletableFuture<ValueClass> putFuture = mapWithCustomSerialization.put(key1, value1);


        putFuture.thenCompose(prevValue -> {
            if (prevValue != null) {
                throw new RuntimeException("Previous value should be null");
            }

            return mapWithCustomSerialization.getAsync(key1).thenAccept(valueFromGet -> {
                if (!Objects.equals(valueFromGet, value1)) {
                    throw new RuntimeException("Values should be equal");
                }

                System.out.println("New value after putAsync: " + valueFromGet);
            });
        }).get(10, TimeUnit.SECONDS);
    }
    }


    private static void basicOperationsSync(IgniteMap<Integer, String> map) {
        map.put(3, "val3");

        String value = map.get(3);

        System.out.println("Value: " + value);

    }

    private static void basicOperationsAsync(IgniteMap<Integer, String> map) throws ExecutionException, InterruptedException, TimeoutException {
        map.putAsync(4, "val4").thenCompose(prev -> {
            if (prev != null) {
                throw new RuntimeException("Previous value should be null");
            }
            return map.getAsync(4)
                    .thenAccept(v -> System.out.println("Value after putAsync: " + v));
        }).get(10, TimeUnit.SECONDS);
    }

    //--------------------------------------------------------------------------------------
    //
    // Key class used to create map2 example.
    //
    //--------------------------------------------------------------------------------------
    private static class KeyClass {
        private final String str;

        private final Integer integer;

        public KeyClass(String s, Integer i) {
            this.str = s;
            this.integer = i;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (!(o instanceof KeyClass)) {
                return false;
            }
            KeyClass keyClass = (KeyClass) o;
            return Objects.equals(str, keyClass.str) && Objects.equals(integer, keyClass.integer);
        }

        @Override
        public int hashCode() {
            return Objects.hash(str, integer);
        }

        @Override
        public String toString() {
            return "KeyClass{" +
                    "str='" + str + '\'' +
                    ", integer=" + integer +
                    '}';
        }
    }

    //--------------------------------------------------------------------------------------
    //
    // Value serializer for the KeyClass.
    //
    //--------------------------------------------------------------------------------------
    private static class KeySerializer implements ByteArrayMarshaller<KeyClass> {
        @Override
        public byte [] marshal(KeyClass object) {
            if (object == null) {
                return null;
            }

            String result = object.integer + ":" + Base64.getEncoder().encodeToString(object.str.getBytes(UTF_8));

            return result.getBytes(UTF_8);
        }

        @Override
        public  KeyClass unmarshal(byte  [] raw) {
            if (raw == null) {
                return null;
            }

            String rawStr = new String(raw, UTF_8);
            int index = rawStr.indexOf(':');

            return new KeyClass(
                    new String(Base64.getDecoder().decode(rawStr.substring(index + 1)), UTF_8),
                    Integer.parseInt(rawStr.substring(0, index))
            );
        }
    }

    //--------------------------------------------------------------------------------------
    //
    // Value class used to create map2 example.
    //
    //--------------------------------------------------------------------------------------
    private static class ValueClass {
        private final List<String> list;

        public ValueClass(List<String> v) {
            this.list = v;
        }

        @Override
        public boolean equals(Object o) {
            if (this == o) {
                return true;
            }
            if (!(o instanceof ValueClass)) {
                return false;
            }
            ValueClass that = (ValueClass) o;
            return Objects.equals(list, that.list);
        }

        @Override
        public int hashCode() {
            return Objects.hashCode(list);
        }

        @Override
        public String toString() {
            return "ValueClass{" +
                    "list=" + list +
                    '}';
        }
    }

    //--------------------------------------------------------------------------------------
    //
    // Value serializer for the ValueClass.
    //
    //--------------------------------------------------------------------------------------
    private static class ValueSerializer implements ByteArrayMarshaller<ValueClass> {
        @Override
        public byte  [] marshal( ValueClass object) {
            if (object == null) {
                return null;
            }

            String result = object.list.stream().map(it -> Base64.getEncoder().encodeToString(it.getBytes(UTF_8)))
                    .collect(Collectors.joining(":"));

            return result.getBytes(UTF_8);
        }


        @Override
        public  ValueClass unmarshal(byte  [] raw) {
            if (raw == null) {
                return null;
            }

            String[] split = new String(raw, UTF_8).split(":");

            return new ValueClass(Arrays.stream(split)
                    .map(it -> new String(Base64.getDecoder().decode(it), UTF_8))
                    .collect(Collectors.toUnmodifiableList())
            );

        }
    }
}