Un trigger en Java que replica la posición actual de un GPS de Traccar (H2) a RethinkDB. Falta limpiar un poco el código, los imports y quitar los strings "mágicos".

RethinkTrigger.java

package mx.org.cloud9.h2_rethink;

import com.rethinkdb.RethinkDB;
import com.rethinkdb.gen.ast.Db;
import com.rethinkdb.gen.ast.Insert;
import com.rethinkdb.gen.ast.Table;
import com.rethinkdb.model.MapObject;
import com.rethinkdb.net.Connection.Builder;
import java.io.PrintStream;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.sql.Timestamp;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.concurrent.TimeoutException;
import org.h2.tools.TriggerAdapter;

public class RethinkTrigger
  extends TriggerAdapter {
  private static final RethinkDB r = RethinkDB.r;
  
  /** Para hacer pruebas desde la terminal **/
  public static void main(String... args)
    throws Exception {
    System.out.println("Testing from shell.");
    
    com.rethinkdb.net.Connection rethink_conn = r.connection().hostname("servername").port(28015).connect();
    
    Class.forName("org.h2.Driver");
    java.sql.Connection conn = DriverManager.getConnection("jdbc:h2:file:/home/eromero/IdeaProjects/h2-rethink/data/database;USER=sa;");
    Statement stat = conn.createStatement();
    
    ResultSet rs = stat.executeQuery("SELECT * FROM POSITIONS LIMIT 1");
    
    ResultSetMetaData metaData = rs.getMetaData();
    int num_cols = metaData.getColumnCount();
    while (rs.next()) {
      MapObject data = r.hashMap();
      for (int i = 1; i <= num_cols; i++) {
        String label = metaData.getColumnLabel(i);
        Object value = rs.getObject(i);
        String field_type = "" + rs.getObject(i).getClass();
        if (field_type.equals("class java.sql.Timestamp")) {
          Timestamp date_value = (Timestamp)rs.getObject(i);
          LocalDateTime localDateTime = LocalDateTime.ofInstant(date_value.toInstant(), ZoneId.systemDefault());
          
          String date_time = localDateTime.getYear() + "-" + String.format("%02d", new Object[] { Integer.valueOf(localDateTime.getMonthValue()) }) + "-" + String.format("%02d", new Object[] { Integer.valueOf(localDateTime.getDayOfMonth()) }) + "T" + String.format("%02d", new Object[] {Integer.valueOf(localDateTime.getHour()) }) + ":" + String.format("%02d", new Object[] { Integer.valueOf(localDateTime.getMinute()) }) + ':' + String.format("%02d", new Object[] { Integer.valueOf(localDateTime.getSecond()) }) + "+00:00";
          value = r.iso8601(date_time);
        }
        data.put(label.toLowerCase(), value);
        System.out.println(label + ": " + value + " -> " + value.getClass());
      }
      data.put("id", data.get("deviceid"));
      data.put("current_location", r.point(data.get("longitude"), data.get("latitude")));
      data.put("updated", r.now());
      
      r.db("geo").table("locations").insert(data).optArg("conflict", "replace").run(rethink_conn);
    }
    stat.close();
    conn.close();
  }
  
  /** Este es el trigger. Se dispara cuando hay un cambio. Recibe la conexión a H2, el Row Original y el nuevo.
   *  Solo estámos interesados en la posición actual (newRow) 
   **/
  public void fire(java.sql.Connection connection, ResultSet oldRow, ResultSet newRow)
    throws SQLException {
    try {
      com.rethinkdb.net.Connection rethink_conn = r.connection().hostname("servername").port(28015).connect();
      
      ResultSetMetaData metaData = newRow.getMetaData();
      int num_cols = metaData.getColumnCount();
      if ((newRow != null) && (newRow.next())) {
        MapObject data = r.hashMap();
        for (int i = 1; i <= num_cols; i++) {
          String label = metaData.getColumnLabel(i);
          Object value = newRow.getObject(i);
          String field_type = "" + newRow.getObject(i).getClass();
          if (field_type.equals("class java.sql.Timestamp")) {
            Timestamp date_value = (Timestamp)newRow.getObject(i);
            LocalDateTime localDateTime = LocalDateTime.ofInstant(date_value.toInstant(), ZoneId.systemDefault());
            
            String date_time = localDateTime.getYear() + "-" + String.format("%02d", new Object[] { Integer.valueOf(localDateTime.getMonthValue()) }) + "-" + String.format("%02d", new Object[] { Integer.valueOf(localDateTime.getDayOfMonth()) }) + "T" + String.format("%02d", new Object[] {Integer.valueOf(localDateTime.getHour()) }) + ":" + String.format("%02d", new Object[] { Integer.valueOf(localDateTime.getMinute()) }) + ':' + String.format("%02d", new Object[] { Integer.valueOf(localDateTime.getSecond()) }) + "+00:00";
            value = r.iso8601(date_time);
          }
          data.put(label.toLowerCase(), value);
        }
        data.put("id", data.get("deviceid"));
        data.put("current_location", r.point(data.get("longitude"), data.get("latitude")));
        data.put("updated", r.now());
        
        r.db("geo").table("locations").insert(data).optArg("conflict", "replace").run(rethink_conn);
      }
      rethink_conn.close();
    } catch (TimeoutException e) {
      System.out.println(e.toString());
    }
  }
}

El trigger es disparado cuando hay un cambio en la base de datos (H2). De newRow obtenemos la información de la posición actual.

Construimos un MapObject para guardar los campos y valores del registro. Tomamos algunas precauciones para convertir el TimeStamp al formato que espera RethinkDB.

RethinkDB recibe el id, la posición actual y el timestamp. En caso de que exista un registro anterior para ese device, es actualizado (Upsert).