diff --git a/src/main/java/com/avaje/ebeaninternal/server/changelog/BulkJsonBuilder.java b/src/main/java/com/avaje/ebeaninternal/server/changelog/ChangeJsonBuilder.java similarity index 58% rename from src/main/java/com/avaje/ebeaninternal/server/changelog/BulkJsonBuilder.java rename to src/main/java/com/avaje/ebeaninternal/server/changelog/ChangeJsonBuilder.java index 18c8bfb02..14c00bcf4 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/changelog/BulkJsonBuilder.java +++ b/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 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. - *

- * Note that for ease of search/use we effectively denormalise by including the transaction header - * information in each bean document. - *

- */ - 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. */ diff --git a/src/main/java/com/avaje/ebeaninternal/server/changelog/DefaultChangeLogListener.java b/src/main/java/com/avaje/ebeaninternal/server/changelog/DefaultChangeLogListener.java new file mode 100644 index 000000000..a0632218e --- /dev/null +++ b/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 org.avaje.ebean.ChangeLog. + *

+ * The logged entries duplicate/denormalise the transaction details so that each bean change + * is fully contained with the transaction information. + *

+ */ +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 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; + } + +} diff --git a/src/main/java/com/avaje/ebeaninternal/server/changelog/ElasticChangeLogListener.java b/src/main/java/com/avaje/ebeaninternal/server/changelog/ElasticChangeLogListener.java deleted file mode 100644 index 08537d29b..000000000 --- a/src/main/java/com/avaje/ebeaninternal/server/changelog/ElasticChangeLogListener.java +++ /dev/null @@ -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; - } - -} diff --git a/src/main/java/com/avaje/ebeaninternal/server/core/InternalConfiguration.java b/src/main/java/com/avaje/ebeaninternal/server/core/InternalConfiguration.java index 23f562a6c..b38230e55 100644 --- a/src/main/java/com/avaje/ebeaninternal/server/core/InternalConfiguration.java +++ b/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()); } /** diff --git a/src/test/java/com/avaje/ebeaninternal/server/changelog/BulkJsonBuilderTest.java b/src/test/java/com/avaje/ebeaninternal/server/changelog/BulkJsonBuilderTest.java deleted file mode 100644 index e38025376..000000000 --- a/src/test/java/com/avaje/ebeaninternal/server/changelog/BulkJsonBuilderTest.java +++ /dev/null @@ -1,30 +0,0 @@ -package com.avaje.ebeaninternal.server.changelog; - -import com.avaje.ebean.BaseTestCase; -import com.avaje.ebean.Ebean; -import com.avaje.ebean.text.json.JsonContext; -import org.junit.Test; - -import java.io.StringWriter; - - -public class BulkJsonBuilderTest extends BaseTestCase { - - Helper helper = new Helper(); - - @Test - public void testToJson() throws Exception { - - JsonContext jsonContext = Ebean.getDefaultServer().json(); - BulkJsonBuilder builder = new BulkJsonBuilder(jsonContext, "changelog2", "changelog2"); - - - StringWriter buffer = new StringWriter(); - builder.writeJson(helper.createChangeSet("ABCD", 10), buffer); - System.out.println(buffer.toString()); - - buffer = new StringWriter(); - builder.writeJson(helper.createChangeSet("ABCD-2", 15), buffer); - System.out.println(buffer.toString()); - } -} \ No newline at end of file diff --git a/src/test/java/com/avaje/ebeaninternal/server/changelog/ChangeJsonBuilderTest.java b/src/test/java/com/avaje/ebeaninternal/server/changelog/ChangeJsonBuilderTest.java new file mode 100644 index 000000000..68319fe4e --- /dev/null +++ b/src/test/java/com/avaje/ebeaninternal/server/changelog/ChangeJsonBuilderTest.java @@ -0,0 +1,33 @@ +package com.avaje.ebeaninternal.server.changelog; + +import com.avaje.ebean.BaseTestCase; +import com.avaje.ebean.Ebean; +import com.avaje.ebean.event.changelog.BeanChange; +import com.avaje.ebean.event.changelog.ChangeSet; +import com.avaje.ebean.text.json.JsonContext; +import org.junit.Test; + +import java.io.StringWriter; +import java.util.List; + + +public class ChangeJsonBuilderTest extends BaseTestCase { + + Helper helper = new Helper(); + + @Test + public void testToJson() throws Exception { + + JsonContext jsonContext = Ebean.getDefaultServer().json(); + ChangeJsonBuilder builder = new ChangeJsonBuilder(jsonContext); + + ChangeSet changeSet = helper.createChangeSet("ABCD", 10); + List changes = changeSet.getChanges(); + for (int i = 0; i < changes.size(); i++) { + StringWriter buffer = new StringWriter(); + builder.writeBeanJson(buffer, changes.get(i), changeSet, i); + System.out.println(buffer.toString()); + } + + } +} \ No newline at end of file diff --git a/src/test/java/com/avaje/ebeaninternal/server/changelog/ElasticChangeLogListenerTest.java b/src/test/java/com/avaje/ebeaninternal/server/changelog/DefaultChangeLogListenerTest.java similarity index 80% rename from src/test/java/com/avaje/ebeaninternal/server/changelog/ElasticChangeLogListenerTest.java rename to src/test/java/com/avaje/ebeaninternal/server/changelog/DefaultChangeLogListenerTest.java index 1ce5ba59f..2030a8d6c 100644 --- a/src/test/java/com/avaje/ebeaninternal/server/changelog/ElasticChangeLogListenerTest.java +++ b/src/test/java/com/avaje/ebeaninternal/server/changelog/DefaultChangeLogListenerTest.java @@ -6,14 +6,14 @@ import com.avaje.ebean.EbeanServer; import com.avaje.ebean.event.changelog.ChangeSet; import org.junit.Test; -public class ElasticChangeLogListenerTest extends BaseTestCase { +public class DefaultChangeLogListenerTest extends BaseTestCase { Helper helper = new Helper(); @Test public void test() { - ElasticChangeLogListener changeLogListener = new ElasticChangeLogListener(); + DefaultChangeLogListener changeLogListener = new DefaultChangeLogListener(); EbeanServer defaultServer = Ebean.getDefaultServer(); changeLogListener.configure(defaultServer.getPluginApi()); diff --git a/src/test/resources/logback-test.xml b/src/test/resources/logback-test.xml index 07bb24d26..b52efc14d 100644 --- a/src/test/resources/logback-test.xml +++ b/src/test/resources/logback-test.xml @@ -22,19 +22,19 @@ - + log/changeLog.log log/changeLog.log.%d{yyyy-MM-dd} 90 - -- %d{HH:mm:ss.SSS} %msg%n + %d{HH:mm:ss.SSS} %msg%n - - + +