meerkat-java/bulletin-board-client/src/main/java/meerkat/bulletinboard/SimpleBulletinBoardClient.java

package meerkat.bulletinboard;

import com.google.protobuf.ByteString;
import com.google.protobuf.Timestamp;
import meerkat.comm.CommunicationException;
import meerkat.comm.MessageInputStream;
import meerkat.crypto.Digest;
import meerkat.crypto.concrete.SHA256Digest;
import meerkat.protobuf.BulletinBoardAPI;
import meerkat.protobuf.BulletinBoardAPI.*;
import meerkat.protobuf.Voting.*;
import meerkat.rest.*;

import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.InvocationTargetException;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;

import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.client.Entity;
import javax.ws.rs.client.WebTarget;
import javax.ws.rs.core.Response;

import static meerkat.bulletinboard.BulletinBoardConstants.*;

/**
 * Created by Arbel Deutsch Peled on 05-Dec-15.
 * Implements BulletinBoardClient interface in a simple, straightforward manner
 */
public class SimpleBulletinBoardClient implements BulletinBoardClient{

    protected List<String> meerkatDBs;

    protected Client client;

    protected Digest digest;

    /**
     * Stores database locations and initializes the web Client
     * @param clientParams contains the data needed to access the DBs
     */
    @Override
    public void init(BulletinBoardClientParams clientParams) {

        this.meerkatDBs = clientParams.getBulletinBoardAddressList();

        client = ClientBuilder.newClient();
        client.register(ProtobufMessageBodyReader.class);
        client.register(ProtobufMessageBodyWriter.class);

        digest = new SHA256Digest();

    }

    /**
     * Post message to all DBs
     * Make only one try per DB.
     * @param msg is the message,
     * @return the message ID for later retrieval
     * @throws CommunicationException
     */
    @Override
    public MessageID postMessage(BulletinBoardMessage msg) throws CommunicationException {

        WebTarget webTarget;
        Response response;

        // Post message to all databases
        try {
            for (String db : meerkatDBs) {
                webTarget = client.target(db).path(BULLETIN_BOARD_SERVER_PATH).path(POST_MESSAGE_PATH);
                response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(msg, Constants.MEDIATYPE_PROTOBUF));

                // Only consider valid responses
                if (response.getStatusInfo() == Response.Status.OK
                        || response.getStatusInfo() == Response.Status.CREATED) {
                    response.readEntity(BoolMsg.class).getValue();
                }
            }
        } catch (Exception e) { // Occurs only when server replies with valid status but invalid data
            throw new CommunicationException("Error accessing database: " + e.getMessage());
        }

        // Calculate the correct message ID and return it
        digest.reset();
        digest.update(msg.getMsg());
        return MessageID.newBuilder().setID(ByteString.copyFrom(digest.digest())).build();
    }

    /**
     * Access each database and search for a given message ID
     * Return the number of databases in which the message was found
     * Only try once per DB
     * Ignore communication exceptions in specific databases
     * @param id is the requested message ID
     * @return the number of DBs in which retrieval was successful
     */
    @Override
    public float getRedundancy(MessageID id) {
        WebTarget webTarget;
        Response response;

        MessageFilterList filterList = MessageFilterList.newBuilder()
                .addFilter(MessageFilter.newBuilder()
                    .setType(FilterType.MSG_ID)
                    .setId(id.getID())
                    .build())
                .build();

        float count = 0;

        for (String db : meerkatDBs) {
            try {
                webTarget = client.target(db).path(BULLETIN_BOARD_SERVER_PATH).path(READ_MESSAGES_PATH);

                response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(filterList, Constants.MEDIATYPE_PROTOBUF));

                if (response.readEntity(BulletinBoardMessageList.class).getMessageCount() > 0){
                    count++;
                }

            } catch (Exception e) {}
        }

        return count / ((float) meerkatDBs.size());
    }

    /**
     * Go through the DBs and try to retrieve messages according to the specified filter
     * If at the operation is successful for some DB: return the results and stop iterating
     * If no operation is successful: return null (NOT blank list)
     * @param filterList return only messages that match the filters (null means no filtering).
     * @return the list of Bulletin Board messages that are returned from a server
     */
    @Override
    public List<BulletinBoardMessage> readMessages(MessageFilterList filterList) {

        WebTarget webTarget;
        Response response;
        BulletinBoardMessageList messageList;

        // Replace null filter list with blank one.
        if (filterList == null){
            filterList = MessageFilterList.newBuilder().build();
        }

        for (String db : meerkatDBs) {

            try {
                webTarget = client.target(db).path(BULLETIN_BOARD_SERVER_PATH).path(READ_MESSAGES_PATH);

                response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(filterList, Constants.MEDIATYPE_PROTOBUF));

                messageList = response.readEntity(BulletinBoardMessageList.class);

                if (messageList != null){
                    return messageList.getMessageList();
                }

            } catch (Exception e) {}

        }

        return null;

    }

    @Override
    public SyncQuery generateSyncQuery(GenerateSyncQueryParams generateSyncQueryParams) throws CommunicationException {

        WebTarget webTarget;
        Response response;

        for (String db : meerkatDBs) {

            try {
                webTarget = client.target(db).path(BULLETIN_BOARD_SERVER_PATH).path(GENERATE_SYNC_QUERY_PATH);

                response = webTarget.request(Constants.MEDIATYPE_PROTOBUF).post(Entity.entity(generateSyncQueryParams, Constants.MEDIATYPE_PROTOBUF));

                return response.readEntity(SyncQuery.class);

            } catch (Exception e) {}

        }

        throw new CommunicationException("Could not contact any server");

    }

    public void close() {
        client.close();
    }

}