arcanechat-android/src/main/java/chat/delta/rpc/BaseTransport.java

/* Autogenerated file, do not edit manually */
package chat.delta.rpc;

import chat.delta.util.SettableFuture;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;

/* Basic RPC Transport implementation */ 
public abstract class BaseTransport implements Rpc.Transport {
  private final Map<Integer, SettableFuture<JsonNode>> requestFutures = new ConcurrentHashMap<>();
  private int requestId = 0;
  private final ObjectMapper mapper = new ObjectMapper();
  private Thread worker;

  /* Send a Request as raw JSON String to the RPC server */
  protected abstract void sendRequest(String jsonRequest);

  /* Get next Response as raw JSON String from the RPC server */
  protected abstract String getResponse();

  public ObjectMapper getObjectMapper() {
    return mapper;
  }

  public void call(String method, JsonNode... params) throws RpcException {
    innerCall(method, params);
  }

  public <T> T callForResult(TypeReference<T> resultType, String method, JsonNode... params) throws RpcException {
    try {
      JsonNode node = innerCall(method, params);
      if (node.isNull()) return null;
      return mapper.readValue(node.traverse(), resultType);
    } catch (IOException e) {
      throw new RpcException(e.getMessage());
    }
  }

  private JsonNode innerCall(String method, JsonNode... params) throws RpcException {
    int id;
    synchronized (this) {
      id = ++requestId;
      ensureWorkerThread();
    }
    try {
      String jsonRequest = mapper.writeValueAsString(new Request(method, params, id));
      SettableFuture<JsonNode> future = new SettableFuture<>();
      requestFutures.put(id, future);
      sendRequest(jsonRequest);
      return future.get();
    } catch (ExecutionException e) {
      throw (RpcException)e.getCause();
    } catch (InterruptedException e) {
      throw new RpcException(e.getMessage());
    } catch (JsonProcessingException e) {
      throw new RpcException(e.getMessage());
    }
  }

  private void ensureWorkerThread() {
    if (worker != null) return;

    worker = new Thread(() -> {
        while (true) {
          try {
            processResponse();
          } catch (JsonProcessingException e) {
            e.printStackTrace();
          }
        }
    }, "jsonrpcThread");
    worker.start();
  }

  private void processResponse() throws JsonProcessingException {
    String jsonResponse = getResponse();
    Response response = mapper.readValue(jsonResponse, Response.class);

    if (response.id == 0) { // Got JSON-RPC notification/event, ignore
      return;
    }

    SettableFuture<JsonNode> future = requestFutures.remove(response.id);
    if (future == null) { // Got a response with unknown ID, ignore
      return;
    }

    if (response.error != null) {
      future.setException(new RpcException(response.error.toString()));
    } else if (response.result != null) {
      future.set(response.result);
    } else {
      future.setException(new RpcException("Got JSON-RPC response without result or error: " + jsonResponse));
    }
  }

  private static class Request {
    private final String jsonrpc = "2.0";
    public final String method;
    public final JsonNode[] params;
    public final int id;

    public Request(String method, JsonNode[] params, int id) {
      this.method = method;
      this.params = params;
      this.id = id;
    }
  }

  private static class Response {
    public String jsonrpc;
    public int id;
    public JsonNode result;
    public JsonNode error;
  }
}