#390 - ChangeLog - Refactor DefaultChangeLogListener (simplify to use standard logger)

src/main/java/com/avaje/ebeaninternal/server/changelog/ChangeJsonBuilder.java

@@ -11,67 +11,42 @@ import com.fasterxml.jackson.core.JsonGenerator;
 
 import java.io.IOException;
 import java.io.Writer;
-import java.util.List;
 import java.util.Map;
 
 /**
- * Builds JSON appropriate for loading into ElasticS via the bulk API.
+ * Builds JSON document for a bean change.
  */
-public class BulkJsonBuilder {
+public class ChangeJsonBuilder {
 
   protected final JsonFactory jsonFactory = new JsonFactory();
 
-  protected final String indexName;
-
-  protected final String indexType;
-
   protected final JsonContext json;
 
-  protected BulkJsonBuilder(JsonContext json, String indexName, String indexType) {
+  protected ChangeJsonBuilder(JsonContext json) {
     this.json = json;
-    this.indexName = indexName;
-    this.indexType = indexType;
   }
 
   /**
-   * Write the change set into Elastic bulk API JSON form (so contains special new line
-   * characters and bulk API header etc.
+   * Write the bean change as JSON.
    */
-  public void writeJson(ChangeSet changeSet, Writer writer) throws IOException {
+  public void writeBeanJson(Writer writer, BeanChange bean, ChangeSet changeSet, int position) throws IOException {
 
     JsonGenerator generator = jsonFactory.createGenerator(writer);
 
-    List<BeanChange> changes = changeSet.getChanges();
-    for (int i = 0; i < changes.size(); i++) {
-      write(generator, changes.get(i), changeSet, i);
-    }
+    writeBeanChange(generator, bean, changeSet, position);
 
     generator.flush();
     generator.close();
   }
 
-  /**
-   * Write the bean change as a single JSON document for storage into Elastic.
-   * <p>
-   * Note that for ease of search/use we effectively denormalise by including the transaction header
-   * information in each bean document.
-   * </p>
-   */
-  protected void write(JsonGenerator gen, BeanChange bean, ChangeSet changeSet, int position) throws IOException {
-
-    writeBulkHeader(gen, changeSet, position);
-    writeBeanChange(gen, bean, changeSet);
-    writeBeanChangeEnd(gen);
-  }
-
   /**
    * Write the bean change as JSON document containing the transaction header details.
    */
-  protected void writeBeanChange(JsonGenerator gen, BeanChange bean, ChangeSet changeSet) throws IOException {
+  protected void writeBeanChange(JsonGenerator gen, BeanChange bean, ChangeSet changeSet, int position) throws IOException {
 
     gen.writeStartObject();
 
-    writeBeanTransactionDetails(gen, changeSet);
+    writeBeanTransactionDetails(gen, changeSet, position);
 
     gen.writeStringField("object", bean.getTable());
     gen.writeStringField("objectId", bean.getId().toString());
@@ -83,21 +58,15 @@ public class BulkJsonBuilder {
     gen.writeEndObject();
   }
 
-  /**
-   * For Elastic bulk we append raw new line character.
-   */
-  protected void writeBeanChangeEnd(JsonGenerator gen) throws IOException {
-    gen.writeRawValue("\n");
-  }
-
   /**
    * Denormalise by writing the transaction header details.
    */
-  protected void writeBeanTransactionDetails(JsonGenerator gen, ChangeSet changeSet) throws IOException {
+  protected void writeBeanTransactionDetails(JsonGenerator gen, ChangeSet changeSet, int position) throws IOException {
 
     gen.writeStringField("txnId", changeSet.getTxnId());
     gen.writeStringField("txnState", changeSet.getTxnState().getCode());
     gen.writeNumberField("txnBatch", changeSet.getTxnBatch());
+    gen.writeNumberField("txnPosition", position);
     gen.writeStringField("userId", changeSet.getUserId());
     String userIpAddress = changeSet.getUserIpAddress();
     if (userIpAddress != null) {
@@ -109,28 +78,6 @@ public class BulkJsonBuilder {
     }
   }
 
-  /**
-   * Write the elastic bulk API header.
-   */
-  protected void writeBulkHeader(JsonGenerator gen, ChangeSet changeSet, int position) throws IOException {
-
-    // we index with an 'id' value so that we can process/reprocess the JSON and
-    // avoid duplicates being inserted. Appending the batch and position give us
-    // an effectively unique id value for the change
-    String uid = changeSet.getTxnId() + "_" + changeSet.getTxnBatch() + "." + position;
-
-    // the 'header' for elastic bulk API
-    gen.writeStartObject();
-    gen.writeFieldName("index");
-    gen.writeStartObject();
-    gen.writeStringField("_index", indexName);
-    gen.writeStringField("_type", indexType);
-    gen.writeStringField("_id", uid);
-    gen.writeEndObject();
-    gen.writeEndObject();
-    gen.writeRawValue("\n");
-  }
-
   /**
    * For insert and update write the new/old values.
    */

src/main/java/com/avaje/ebeaninternal/server/changelog/DefaultChangeLogListener.java

@@ -0,0 +1,103 @@
+package com.avaje.ebeaninternal.server.changelog;
+
+import com.avaje.ebean.event.changelog.BeanChange;
+import com.avaje.ebean.event.changelog.ChangeLogListener;
+import com.avaje.ebean.event.changelog.ChangeSet;
+import com.avaje.ebean.event.changelog.ChangeType;
+import com.avaje.ebean.plugin.SpiServer;
+import com.avaje.ebean.plugin.SpiServerPlugin;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.StringWriter;
+import java.util.List;
+import java.util.Properties;
+
+/**
+ * Logs the change sets in JSON to logger named <code>org.avaje.ebean.ChangeLog</code>.
+ * <p>
+ * The logged entries duplicate/denormalise the transaction details so that each bean change
+ * is fully contained with the transaction information.
+ * </p>
+ */
+public class DefaultChangeLogListener implements ChangeLogListener, SpiServerPlugin {
+
+  /**
+   * The usual application specific logger.
+   */
+  protected static final Logger logger = LoggerFactory.getLogger(DefaultChangeLogListener.class);
+
+  /**
+   * The named logger we send the change set payload to. Can be externally configured as desired.
+   */
+  protected static final Logger changeLog = LoggerFactory.getLogger("org.avaje.ebean.ChangeLog");
+
+  /**
+   * Used to build the JSON.
+   */
+  protected ChangeJsonBuilder jsonBuilder;
+
+  /**
+   * A bigger default buffer for bean inserts and updates (that have value pairs).
+   */
+  protected int defaultBufferSize = 400;
+
+  /**
+   * Expected to be a reasonable buffer size for deletes (which do not have value pairs).
+   */
+  protected int defaultDeleteBufferSize = 250;
+
+  public DefaultChangeLogListener() {
+  }
+
+  /**
+   * Configure the underlying JSON handler.
+   */
+  @Override
+  public void configure(SpiServer server) {
+    jsonBuilder = new ChangeJsonBuilder(server.json());
+
+    Properties properties = server.getServerConfig().getProperties();
+    String bufferSize = properties.getProperty("ebean.changeLog.bufferSize");
+    if (bufferSize != null) {
+      defaultBufferSize = Integer.parseInt(bufferSize);
+    }
+  }
+
+  @Override
+  public void online(boolean online) {
+    // nothing to do
+  }
+
+  @Override
+  public void shutdown() {
+    // nothing to do
+  }
+
+  @Override
+  public void log(ChangeSet changeSet) {
+
+    try {
+      List<BeanChange> changes = changeSet.getChanges();
+      for (int i = 0; i < changes.size(); i++) {
+        // log each bean change as a separate log entry
+        BeanChange beanChange = changes.get(i);
+        StringWriter writer = new StringWriter(getBufferSize(beanChange));
+        jsonBuilder.writeBeanJson(writer, beanChange, changeSet, i);
+        changeLog.info(writer.toString());
+      }
+    } catch (IOException e) {
+      logger.error("Exception sending changeSet " + changeSet.toString(), e);
+    }
+  }
+
+  /**
+   * Return a decent buffer size based on the bean change.
+   */
+  protected int getBufferSize(BeanChange beanChange) {
+
+    return ChangeType.DELETE == beanChange.getType() ? defaultDeleteBufferSize : defaultBufferSize;
+  }
+
+}

src/main/java/com/avaje/ebeaninternal/server/changelog/ElasticChangeLogListener.java

@@ -1,80 +0,0 @@
-package com.avaje.ebeaninternal.server.changelog;
-
-import com.avaje.ebean.event.changelog.ChangeLogListener;
-import com.avaje.ebean.event.changelog.ChangeSet;
-import com.avaje.ebean.plugin.SpiServer;
-import com.avaje.ebean.plugin.SpiServerPlugin;
-import org.jetbrains.annotations.NotNull;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.StringWriter;
-
-/**
- * Logs the change sets in a Elastic Search Bulk API JSON format.
- */
-public class ElasticChangeLogListener implements ChangeLogListener, SpiServerPlugin {
-
-  protected static final Logger appLog = LoggerFactory.getLogger(ElasticChangeLogListener.class);
-
-  protected static final Logger fileLog = LoggerFactory.getLogger("org.avaje.ebean.ElasticChangeLog");
-
-  protected BulkJsonBuilder jsonBuilder;
-
-  public ElasticChangeLogListener() {
-  }
-
-  /**
-   * Configure the underlying JSON handler.
-   */
-  @Override
-  public void configure(SpiServer server) {
-    jsonBuilder = new BulkJsonBuilder(server.json(), "changelog", "changelog");
-  }
-
-  @Override
-  public void online(boolean online) {
-    // We don't care online or offline in this case
-    // we might if we setup for network sending etc
-  }
-
-  @Override
-  public void log(ChangeSet changeSet) {
-
-    try {
-      // I'm pretty sure I'm going to change this to use a  FileWriter (but apache kafta could be good here too)
-      // This buffer could get really big and to me normal logging (without a Writer) is not that well suited to
-      // this problem so ... works but lets do better here and write direct to files (without the buffer issue)
-      StringWriter writer = new StringWriter(getBufferSize(changeSet));
-      jsonBuilder.writeJson(changeSet, writer);
-      String json = writer.toString();
-
-      fileLog.info("Sending txnId:{} txnState:{} txnBatch:{}  \n {}", changeSet.getTxnId(), changeSet.getTxnState(), changeSet.getTxnBatch(), json);
-
-    } catch (IOException e) {
-      String msg = extractErrorMessage(e);
-      fileLog.error("Exception sending txnId:{} txnState:{} txnBatch:{} error:{}", changeSet.getTxnId(), changeSet.getTxnState(), changeSet.getTxnBatch(), msg);
-      appLog.error("Exception sending changeSet "+changeSet.toString(), e);
-    }
-  }
-
-  protected int getBufferSize(ChangeSet changeSet) {
-    // a rough guess, could get smarter here
-    return Math.min(400 * changeSet.size(), 3000);
-  }
-
-  /**
-   * Extract an error message that does not have new line characters and hence safe to go into
-   * our log which contains the payloads (that we want to be able to extract easily later on).
-   */
-  @NotNull
-  protected String extractErrorMessage(Exception e) {
-
-    String msg = e.toString();
-    msg = msg.replace('\r','|');
-    msg = msg.replace('\n', '|');
-    return msg;
-  }
-
-}

src/main/java/com/avaje/ebeaninternal/server/core/InternalConfiguration.java

@@ -17,7 +17,7 @@ import com.avaje.ebeaninternal.server.autofetch.AutoFetchManager;
 import com.avaje.ebeaninternal.server.autofetch.AutoFetchManagerFactory;
 import com.avaje.ebeaninternal.server.changelog.DefaultChangeLogPrepare;
 import com.avaje.ebeaninternal.server.changelog.DefaultChangeLogRegister;
-import com.avaje.ebeaninternal.server.changelog.ElasticChangeLogListener;
+import com.avaje.ebeaninternal.server.changelog.DefaultChangeLogListener;
 import com.avaje.ebeaninternal.server.cluster.ClusterManager;
 import com.avaje.ebeaninternal.server.deploy.BeanDescriptorManager;
 import com.avaje.ebeaninternal.server.deploy.DeployOrmXml;
@@ -187,7 +187,7 @@ public class InternalConfiguration {
    * Return the ChangeLogListener to use with a default implementation if none defined.
    */
   public ChangeLogListener changeLogListener(ChangeLogListener listener) {
-    return plugin((listener != null) ? listener : new ElasticChangeLogListener());
+    return plugin((listener != null) ? listener : new DefaultChangeLogListener());
   }
 
   /**