Adaptor Guide for Java Developers

public class Main {

    private static boolean shutdown = false;
    private static final Object lock = new Object();
    private static final ExampleService service;

    public static void main(String[] args) throws Exception {
        synchronized (lock) {
            Runtime.getRuntime().addShutdownHook(new Thread(QuickStartAdaptorApplication::shutdown));

            service = new ExampleService();

            // wait for a shutdown request to exit
            while (!shutdown) {
                lock.wait();
            }
        }
    }

    private static void shutdown() {
        synchronized (lock) {
            shutdown = true;
            if (service != null)
                service.close();
            lock.notifyAll();
        }
    }
}

package us.younite.adaptor;

import us.younite.adaptor.sdk.builder.AdaptorServiceBuilder;
import us.younite.adaptor.sdk.listener.ds.AbstractDatasource;
import us.younite.adaptor.sdk.listener.ds.Datasource;
import us.younite.adaptor.sdk.model.DataEventResponse;
import us.younite.adaptor.sdk.model.DataRequest;
import us.younite.adaptor.sdk.model.DataResponse;
import us.younite.adaptor.sdk.model.IncomingDataEvent;
import us.younite.adaptor.sdk.model.id.LocalIdentifierFactory;
import us.younite.adaptor.sdk.model.id.simple.LongIdentifier;
import us.younite.adaptor.sdk.service.AdaptorService;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import javax.annotation.ParametersAreNonnullByDefault;
import java.io.Closeable;
import java.util.Optional;
import java.util.UUID;

/**
 * This class is used in the YOUnite-Adaptor-Guide-for-Java-Developers example.
 */
@ParametersAreNonnullByDefault
public class AdaptorGuideExample implements Closeable {

    private final AdaptorService adaptorService;

    public AdaptorGuideExample() throws Exception {
        //
        // get configuration from the environment
        //
        String adaptorUuid = System.getenv("ADAPTOR_UUID");
        String clientId = System.getenv("ADAPTOR_CLIENT_ID");
        String clientSecret = System.getenv("ADAPTOR_CLIENT_SECRET");
        String messageBusUrl = System.getenv("MESSAGE_BUS_URL");
        String messageBusConcurrency = System.getenv("MESSAGE_BUS_CONCURRENCY");
        String messageBusOpsConcurrency = System.getenv("MESSAGE_BUS_OPS_CONCURRENCY");
        String messageBusAssemblyConcurrency = System.getenv("MESSAGE_BUS_ASSEMBLY_CONCURRENCY");

        // records in the data source will be identified by a long, for example,
        // a database table with an auto-incrementing primary key.
        LocalIdentifierFactory<LongIdentifier> localIdentifierFactory = LongIdentifier.getFactory();

        // create an AbstractDatasource implementation
        Datasource<LongIdentifier> datasource = new AbstractDatasource<LongIdentifier>("person", 1, localIdentifierFactory) {
            @Nonnull
            @Override
            public Optional<DataResponse<LongIdentifier>> get(DataRequest<LongIdentifier> request, @Nullable IncomingDataEvent<LongIdentifier> dataEvent, @Nullable DataEventResponse<LongIdentifier> response) {
                // TODO - get a record from the data source
                return Optional.empty();
            }

            @Nonnull
            @Override
            public DataEventResponse<LongIdentifier> put(IncomingDataEvent<LongIdentifier> incoming) {
                // TODO - add or update a record in the data source and return its local identifier
                return DataEventResponse.failure("TODO");
            }

            @Nonnull
            @Override
            public DataEventResponse<LongIdentifier> remove(IncomingDataEvent<LongIdentifier> incoming) {
                // TODO - remove a record in the data source
                return DataEventResponse.failure("TODO");
            }
        };

        // build the adaptor service
        adaptorService = new AdaptorServiceBuilder()
                .uuid(UUID.fromString(adaptorUuid))
                .clientId(clientId)
                .clientSecret(clientSecret)
                .messageBusUrl(messageBusUrl)
                .concurrency(Integer.parseInt(messageBusConcurrency))
                .opsConcurrency(messageBusOpsConcurrency)
                .assemblyConcurrency(messageBusAssemblyConcurrency)
                .datasource(datasource)
                .build();

        // start the adaptor service
        adaptorService.start(60);
    }

    @Override
    public void close() {
        adaptorService.close();
    }
}

package us.younite.adaptor;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import us.younite.adaptor.sdk.builder.AdaptorServiceBuilder;
import us.younite.adaptor.sdk.exceptions.AdaptorConfigurationException;
import us.younite.adaptor.sdk.exceptions.AdaptorStartupException;
import us.younite.adaptor.sdk.listener.change.Change;
import us.younite.adaptor.sdk.listener.ds.InMemDatabase;
import us.younite.adaptor.sdk.model.id.simple.LongIdentifier;
import us.younite.adaptor.sdk.service.AdaptorService;

import javax.annotation.ParametersAreNonnullByDefault;
import java.io.Closeable;
import java.util.UUID;

/**
 * * This is a quick start, referenced in the Adaptor Guide for Java Developers.
 *
 * <p>This application creates two adaptors that use test databases and simple listeners.
 * Data inserted, updated or deleted in the database of one adaptor will be synchronized
 * with the other adaptor.
 *
 * @see InMemDatabase
 * @author zeke@younite.us
 */
@ParametersAreNonnullByDefault
public class QuickStartAdaptorApplication {

    private static final Logger logger = LoggerFactory.getLogger(QuickStartAdaptorApplication.class);
    private static final ObjectMapper objectMapper = new ObjectMapper();

    //
    // BEFORE RUNNING, CREATE THE TEST DOMAIN AND ADAPTORS!!!
    //
    // 1) Create a domain named person with three fields and the exact matching type:
    // a. personId (string) - set as the DR Key property
    // b. name (string)
    // c. phone (string)
    //
    // 2) Create two adaptors and give them all capabilities to the person domain.
    // Enter the adaptors credentials below.
    //
    private static final String messageBusUrl = "tcp://localhost:61616";
    private static final UUID adaptorUuid_1 = UUID.fromString("46193a67-5243-483f-a129-85b5620c0574");
    private static final String clientId_1 = "b84a3379-02e8-4b17-b05e-24ffb3caec18";
    private static final String clientSecret_1 = "ea6e038d-dc5e-437d-85c1-ec094003996a";
    private static final UUID adaptorUuid_2 = UUID.fromString("a76e67f9-5aeb-47e6-a2d8-cbd60ef4dd32");
    private static final String clientId_2 = "a8bb9a19-cb49-4a2f-a730-ee3c2e7d7ef6";
    private static final String clientSecret_2 = "35e5fb86-6d97-4aa9-a774-ebd9d1d49984";
    //
    // UPDATE THE VALUES ABOVE BEFORE RUNNING!
    //

    private static boolean shutdown = false;
    private static final Object lock = new Object();

    public static void main(String[] args) throws Exception {
        synchronized (lock) {
            Runtime.getRuntime().addShutdownHook(new Thread(QuickStartAdaptorApplication::shutdown));

            // adaptor #1 uses the polling method to look for changes
            TestAdaptor adaptor1 = new TestAdaptor(adaptorUuid_1, clientId_1, clientSecret_1, true);

            // adaptor #2 uses an event-driven model of sending changes to YOUnite
            TestAdaptor adaptor2 = new TestAdaptor(adaptorUuid_2, clientId_2, clientSecret_2, false);

            // create some test records
            ObjectNode rylan = testRecord("BCN487", "Rylan Jorgenson", "555-202-4571");
            ObjectNode juneau = testRecord("ABA055", "Juneau Momo", "555-202-8644");
            ObjectNode jobe = testRecord("VRQ721", "Jobe Fallby", "555-230-9732");
            ObjectNode nevel = testRecord("PRX002", "Nevel Andersen", "555-230-8877");

            // add the data to adaptor 1
            logger.info("Creating test data in adaptor 1...");
            adaptor1.database.change(Change.Type.NEW, rylan);
            adaptor1.database.change(Change.Type.NEW, juneau);
            adaptor1.database.change(Change.Type.NEW, jobe);
            adaptor1.database.change(Change.Type.NEW, nevel);
            logger.info("Contents of adaptor 1's database {}", adaptor1.database.toMap());

            // wait for the data to show up in adaptor 2
            while (!shutdown && adaptor2.database.size() < 4) {
                logger.info("Waiting for changes to show up in adaptor 2 ...");
                lock.wait(1000);
            }

            logger.info("Contents of adaptor2: {}", adaptor2.database.toMap());

            // make sure the values are identical
            if (adaptor1.database.size() == adaptor2.database.size() &&
                    CollectionUtils.containsAll(adaptor1.database.toMap().values(), adaptor2.database.toMap().values())) {
                logger.info("SUCCESS! Values are identical in both databases");
            } else {
                logger.error("ERROR! Values are not identical!");
            }

            // change data in adaptor 2
            logger.info("Changing Rylan phone number in adaptor 2 ...");
            rylan.put("phone", "555-233-1144");
            adaptor2.database.change(Change.Type.UPDATE, rylan);

            // wait for the change to show up in adaptor 1
            while (!shutdown) {
                LongIdentifier id = adaptor1.database.lookup(rylan);
                if (id == null) {
                    logger.error("ERROR! Unable to determine id for Rylan record in adaptor 1!");
                    break;
                }

                ObjectNode updated = adaptor1.database.get(id);

                if (updated != null && rylan.get("phone").equals(updated.get("phone"))) {
                    logger.info("Success, found updated record in adaptor 1");
                    break;
                }

                logger.info("Waiting for update to show up in adaptor 1 ...");
                lock.wait(1000);
            }

            // delete all the records in adaptor 1
            for (LongIdentifier i : adaptor1.database.toMap().keySet()) {
                adaptor1.database.change(Change.Type.DELETE, i, null);
            }

            // wait for the deletions to show up in adaptor 2
            while (!shutdown && adaptor2.database.size() > 0) {
                logger.info("Waiting for deletions to show up in adaptor 2 ...");
                lock.wait(1000);
            }

            if (adaptor1.database.size() == 0 && adaptor2.database.size() == 0) {
                logger.info("SUCCESS! All records have been deleted in both databases");
            } else {
                logger.error("ERROR! Not all records have been deleted");
            }

            logger.info("Shutting down adaptors ...");

            adaptor1.close();
            adaptor2.close();
        }
    }

    private static void shutdown() {
        synchronized (lock) {
            shutdown = true;
            lock.notifyAll();
        }
    }

    static class TestAdaptor implements Closeable {
        private final AdaptorService adaptorService;
        private final InMemDatabase database;

        TestAdaptor(UUID adaptorUuid, String clientId, String clientSecret, boolean polling)
                throws AdaptorConfigurationException, AdaptorStartupException, InterruptedException  {

            // Create a test database for this domain version which stores data in memory. Records in the
            // database are uniquely identified locally by an auto-incrementing long integer and indexed by
            // primary key (personId).
            database = InMemDatabase.newPrimaryKeyDatabase("person", 1, "personId", polling);

            adaptorService = new AdaptorServiceBuilder()
                    .uuid(adaptorUuid)
                    .clientId(clientId)
                    .clientSecret(clientSecret)
                    .messageBusUrl(messageBusUrl)
                    .concurrency(4)
                    .datasource(database)
                    .build();

            adaptorService.start(60);
        }

        @Override
        public void close() {
            adaptorService.close();
        }
    }

    private static ObjectNode testRecord(String personId, String name, String phone) {
        ObjectNode node = objectMapper.createObjectNode();
        node.put("personId", personId);
        node.put("name", name);
        node.put("phone", phone);
        return node;
    }
}

log-entry-type: "ADAPTOR_ERROR"

log-entry-type: "ADAPTOR_ERROR" AND source-adaptor-uuid: "8862fdc0-43c9-4ec6-b5b0-a14ec3bdd352"