From 3a24ba832bc82e87006c29a0a16506e3071b7071 Mon Sep 17 00:00:00 2001 From: James Wickham Date: Wed, 27 Sep 2023 10:49:42 +0100 Subject: [PATCH 01/10] renamed and moved classes to a more appropriate package. --- ...StoreEntry.java => JdbcRetryRowEntry.java} | 0 .../core/jdbc/retry/RetryServiceImp.java | 165 ------------- .../jdbc/retry/JdbcAcknowledgeService.java} | 19 +- .../jdbc/retry/JdbcRetryMessagesService.java} | 30 +-- .../jdbc/retry/JdbcRetryServiceImp.java} | 233 ++++++++++-------- .../JdbcStoreMessageForRetryService.java} | 21 +- .../core/jdbc/retry/JdbcRetryStoreTest.java | 28 --- .../retry/JdbcAcknowledgeServiceTest.java | 14 ++ .../retry/JdbcRetryMessagesServiceTest.java | 14 ++ .../JdbcStoreMessageForRetryServiceTest.java | 14 ++ 10 files changed, 209 insertions(+), 329 deletions(-) rename interlok-core/src/main/java/com/adaptris/core/jdbc/retry/{JdbcRetryStoreEntry.java => JdbcRetryRowEntry.java} (100%) delete mode 100644 interlok-core/src/main/java/com/adaptris/core/jdbc/retry/RetryServiceImp.java rename interlok-core/src/main/java/com/adaptris/core/{jdbc/retry/AcknowledgeService.java => services/jdbc/retry/JdbcAcknowledgeService.java} (70%) rename interlok-core/src/main/java/com/adaptris/core/{jdbc/retry/RetryMessagesService.java => services/jdbc/retry/JdbcRetryMessagesService.java} (86%) rename interlok-core/src/main/java/com/adaptris/core/{jdbc/retry/JdbcRetryStore.java => services/jdbc/retry/JdbcRetryServiceImp.java} (70%) rename interlok-core/src/main/java/com/adaptris/core/{jdbc/retry/StoreMessageForRetryService.java => services/jdbc/retry/JdbcStoreMessageForRetryService.java} (92%) delete mode 100644 interlok-core/src/test/java/com/adaptris/core/jdbc/retry/JdbcRetryStoreTest.java create mode 100644 interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeServiceTest.java create mode 100644 interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesServiceTest.java create mode 100644 interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryServiceTest.java diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryStoreEntry.java b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java similarity index 100% rename from interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryStoreEntry.java rename to interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/RetryServiceImp.java b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/RetryServiceImp.java deleted file mode 100644 index 6a73e3f78..000000000 --- a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/RetryServiceImp.java +++ /dev/null @@ -1,165 +0,0 @@ -package com.adaptris.core.jdbc.retry; - -import java.sql.SQLException; - -import javax.validation.constraints.NotNull; - -import org.apache.commons.lang3.BooleanUtils; - -import com.adaptris.annotation.InputFieldDefault; -import com.adaptris.core.AdaptrisConnection; -import com.adaptris.core.AdaptrisMarshaller; -import com.adaptris.core.AdaptrisMessage; -import com.adaptris.core.CoreException; -import com.adaptris.core.DefaultMarshaller; -import com.adaptris.core.ServiceException; -import com.adaptris.core.http.jetty.retry.RetryStore; -import com.adaptris.core.jdbc.DatabaseConnection; -import com.adaptris.core.jdbc.JdbcService; -import com.adaptris.core.util.ExceptionHelper; - -import lombok.Getter; -import lombok.NoArgsConstructor; -import lombok.NonNull; -import lombok.Setter; - - -/** - *

- * Partial implementation of behaviour common to retry services. - *

- */ -@NoArgsConstructor -public abstract class RetryServiceImp extends JdbcService { - - protected static AdaptrisMarshaller marshaller; - - static { // only create marshaller once - marshaller = DefaultMarshaller.getDefaultMarshaller(); - } - - @Getter - @Setter - private AdaptrisConnection connection; - - @NotNull - @NonNull - private RetryStore retryStore; - - @InputFieldDefault(value = "true") - private boolean pruneAcknowledged; - - /** - *

- * Creates a new instance. Defaults to JdbcRetryStore. - *

- */ - - - /** @see com.adaptris.core.AdaptrisComponent#init() */ - @Override - protected void initJdbcService() throws CoreException { - if (getConnection() == null) { - throw new CoreException("DatabaseConnection is null in service"); - } - - getConnection().init(); - getRetryStore().makeConnection(connection); - getRetryStore().init(); - } - - /** @see com.adaptris.core.ServiceImp#start() */ - @Override - public void start() throws CoreException { - getConnection().start(); - } - - /** @see com.adaptris.core.AdaptrisComponent#close() */ - @Override - protected void closeJdbcService() { - getConnection().close(); - } - - public final void doService(AdaptrisMessage msg) throws ServiceException { - pruneAcknowledged(); - performService(msg); - } - - protected abstract void performService(AdaptrisMessage msg) - throws ServiceException; - - private void pruneAcknowledged() { - try { - if (isPruneAcknowledged()) { - log.debug("Pruning Previously Acknowledged Messages"); - getRetryStore().deleteAcknowledged(); - } - } - catch (Exception e) { - log.warn("Ignoring exception while pruning acknowledged messages[" - + e.getMessage() + "]"); - } - } - - // properties - - /** - *

- * Returns the RetryStore to use. - *

- * - * @return the RetryStore to use - */ - public final RetryStore getRetryStore() { - return retryStore; - } - - /** - *

- * Sets the RetryStore to use. May not be null. - *

- * - * @param r the RetryStore to use - */ - public final void setRetryStore(RetryStore r) { - if (r == null) { - throw new IllegalArgumentException("null param"); - } - this.retryStore = r; - } - - - /** - * @return the pruneAcknowledged - */ - public boolean getPruneAcknowledged() { - return pruneAcknowledged; - } - - /** - * Specify whether to delete messages from the underlying store if they have - * already been acknowledged. - * - * @param b the pruneAcknowledged to set - */ - public void setPruneAcknowledged(boolean b) { - this.pruneAcknowledged = b; - } - - private boolean isPruneAcknowledged() { - return BooleanUtils.toBooleanDefaultIfNull(getPruneAcknowledged(), false); - } - - @Override - protected void prepareService() throws CoreException { - // TODO Auto-generated method stub - - } - - @Override - protected void startService() throws CoreException { - // TODO Auto-generated method stub - - } - -} diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/AcknowledgeService.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeService.java similarity index 70% rename from interlok-core/src/main/java/com/adaptris/core/jdbc/retry/AcknowledgeService.java rename to interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeService.java index 88686ce00..277c8ddb4 100644 --- a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/AcknowledgeService.java +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeService.java @@ -1,36 +1,37 @@ -package com.adaptris.core.jdbc.retry; +package com.adaptris.core.services.jdbc.retry; import com.adaptris.annotation.AdapterComponent; import com.adaptris.annotation.ComponentProfile; import com.adaptris.annotation.DisplayOrder; import com.adaptris.core.AdaptrisMessage; import com.adaptris.core.ServiceException; +import com.adaptris.core.jdbc.retry.Constants; import com.adaptris.interlok.InterlokException; import com.thoughtworks.xstream.annotations.XStreamAlias; /** *

* Service which processes asynchronous acknowledgements for messages stored - * using {@link StoreMessageForRetryServiceTest}. + * using {@link JdbcStoreMessageForRetryService}. *

*

* The following metadata keys are required. *

*

*/ -@XStreamAlias("acknowledge-message-service") +@XStreamAlias("jdbc-acknowledge-message-service") @AdapterComponent -@ComponentProfile(summary = "processes asynchronous acknowledgements.", since = "4.9.0", tag = "retry") +@ComponentProfile(summary = "processes asynchronous acknowledgements.", since = "5.0.0", tag = "jdbc, retry") @DisplayOrder(order = { "pruneExpired", "retryStore" }) -public class AcknowledgeService extends RetryServiceImp { +public class JdbcAcknowledgeService extends JdbcRetryServiceImp { /** * - * @see RetryServiceImpTest#performService(com.adaptris.core.AdaptrisMessage) + * @see JdbcRetryServiceImp#performService(com.adaptris.core.AdaptrisMessage) */ @Override protected void performService(AdaptrisMessage msg) throws ServiceException { @@ -42,7 +43,7 @@ protected void performService(AdaptrisMessage msg) throws ServiceException { } try { log.debug("Acknowledging [" + acknowledgeId + "] as successfully sent"); - getRetryStore().acknowledge(acknowledgeId); + acknowledge(acknowledgeId); } catch (InterlokException e) { throw new ServiceException(e); } diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/RetryMessagesService.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesService.java similarity index 86% rename from interlok-core/src/main/java/com/adaptris/core/jdbc/retry/RetryMessagesService.java rename to interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesService.java index 4ea634faa..0ad8ca5f3 100644 --- a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/RetryMessagesService.java +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesService.java @@ -1,4 +1,4 @@ -package com.adaptris.core.jdbc.retry; +package com.adaptris.core.services.jdbc.retry; import java.util.List; @@ -16,12 +16,13 @@ import com.adaptris.core.Service; import com.adaptris.core.ServiceException; import com.adaptris.core.StandaloneProducer; +import com.adaptris.core.jdbc.retry.Constants; import com.adaptris.interlok.InterlokException; import com.thoughtworks.xstream.annotations.XStreamAlias; /** *

- * Service which obtains messages from the retry store that meet the appropriate + * Service which obtains messages from the Database that meet the appropriate * criteria and retries them. This service is intended to be used in conjunction * with PollingTrigger. *

@@ -35,12 +36,11 @@ *

*/ -@XStreamAlias("retry-message-service") +@XStreamAlias("jdbc-retry-message-service") @AdapterComponent -@ComponentProfile(summary = "retries a message from the retry store.", - since = "4.9.0", tag = "retry") +@ComponentProfile(summary = "retries a message from the retry store.", since = "5.0.0", tag = "jdbc, retry") @DisplayOrder(order = {"pruneExpired", "retryStore"}) -public class RetryMessagesService extends RetryServiceImp { +public class JdbcRetryMessagesService extends JdbcRetryServiceImp { @InputFieldDefault(value = "false") private boolean pruneExpired; @@ -49,7 +49,7 @@ public class RetryMessagesService extends RetryServiceImp { @Valid private StandaloneProducer expiredMessagesProducer; - public RetryMessagesService() { + public JdbcRetryMessagesService() { setExpiredMessagesProducer(new StandaloneProducer()); setPruneExpired(false); } @@ -77,13 +77,13 @@ protected void stopService() { /** * - * @see RetryServiceImpTest#performService(com.adaptris.core.AdaptrisMessage) + * @see JdbcRetryServiceImp#performService(com.adaptris.core.AdaptrisMessage) */ @Override protected void performService(AdaptrisMessage msg) throws ServiceException { try { pruneExpired(); - List retryMsgs = getRetryStore().obtainMessagesToRetry(); + List retryMsgs = obtainMessagesToRetry(); for (AdaptrisMessage retry : retryMsgs) { doRetry(retry); } @@ -135,11 +135,11 @@ private void handleSynchronous(AdaptrisMessage retry, Service service) try { service.doService(retry); - getRetryStore().acknowledge( + acknowledge( retry.getMetadataValue(Constants.ACKNOWLEDGE_ID_KEY)); } catch (InterlokException e) { - getRetryStore().updateRetryCount(retry.getUniqueId()); + updateRetryCount(retry.getUniqueId()); } } @@ -147,12 +147,12 @@ private void handleAsynchronous(AdaptrisMessage retry, Service service) throws InterlokException { try { service.doService(retry); - getRetryStore().updateRetryCount(retry.getUniqueId()); + updateRetryCount(retry.getUniqueId()); } catch (ServiceException e) { if ("true".equalsIgnoreCase(retry .getMetadataValue(Constants.ASYNC_AUTO_RETRY))) { - getRetryStore().updateRetryCount(retry.getUniqueId()); + updateRetryCount(retry.getUniqueId()); } else { throw e; @@ -164,12 +164,12 @@ private void pruneExpired() { try { if (isPruneExpired()) { log.debug("Pruning Expired Messages"); - List expiredMsgs = getRetryStore().obtainExpiredMessages(); + List expiredMsgs = obtainExpiredMessages(); for (AdaptrisMessage expired : expiredMsgs) { log.debug("Producing Expired Message " + expired.getUniqueId()); log.debug("EXPIRED MESSAGE" + expired.toString()); getExpiredMessagesProducer().produce(expired); - getRetryStore().delete(expired.getUniqueId()); + delete(expired.getUniqueId()); } } } diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryStore.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceImp.java similarity index 70% rename from interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryStore.java rename to interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceImp.java index 8a071a17c..37310fb5b 100644 --- a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryStore.java +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceImp.java @@ -1,4 +1,4 @@ -package com.adaptris.core.jdbc.retry; +package com.adaptris.core.services.jdbc.retry; import java.io.FileInputStream; import java.io.InputStream; @@ -7,49 +7,47 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; -import java.util.Collections; import java.util.Date; import java.util.List; -import java.util.Map; import java.util.Properties; import javax.validation.constraints.NotBlank; -import com.adaptris.annotation.ComponentProfile; -import com.adaptris.annotation.DisplayOrder; +import org.apache.commons.lang3.BooleanUtils; + import com.adaptris.annotation.InputFieldDefault; -import com.adaptris.core.AdaptrisConnection; +import com.adaptris.core.AdaptrisMarshaller; import com.adaptris.core.AdaptrisMessage; -import com.adaptris.core.AdaptrisMessageFactory; import com.adaptris.core.CoreException; +import com.adaptris.core.DefaultMarshaller; import com.adaptris.core.MimeEncoder; -import com.adaptris.core.http.jetty.retry.RetryStore; +import com.adaptris.core.ServiceException; import com.adaptris.core.jdbc.DatabaseConnection; +import com.adaptris.core.jdbc.JdbcService; +import com.adaptris.core.jdbc.retry.Constants; +import com.adaptris.core.jdbc.retry.JdbcRetryRowEntry; import com.adaptris.core.util.ExceptionHelper; import com.adaptris.core.util.JdbcUtil; import com.adaptris.interlok.InterlokException; -import com.adaptris.interlok.cloud.RemoteBlob; import com.adaptris.interlok.util.Args; -import com.thoughtworks.xstream.annotations.XStreamAlias; - import lombok.Getter; +import lombok.NoArgsConstructor; import lombok.Setter; -import lombok.extern.slf4j.Slf4j; + /** *

- * JDBC-based implementation of RetryStore. + * JDBC services for storing and retrying messages with acknowledgment support. *

*

- * This uses JdbcTemplate from Spring for database operations. If - * there is no explicit configuration of the sqlProperties then the SQL + * If there is no explicit configuration of the sqlProperties then the SQL * statements are found in the file retry-store-derby.properties which * is suitable for Apache Derby and MySQL. If explicitly configured, the * property file is expected to be present on the classpath. *

*

- * The default property file is listed below. + * The default property file statements are listed below. * * 


  create.sql = CREATE TABLE retry_store \
@@ -81,7 +79,7 @@ CONSTRAINT idx_acknowledge_id UNIQUE (acknowledge_id))
  updated_on=? WHERE message_id=?
 
  retry.sql = SELECT * FROM retry_store WHERE \
- (acknowledged='F' AND (retries_to_date < total_retries OR total_retries = -1))
+ (acknowledged='F' AND (retries_to_date < total_retries OR total_retries = -1))
 
  delete.acknowleged.sql = DELETE FROM retry_store WHERE acknowledged='T'
 
@@ -89,7 +87,7 @@ CONSTRAINT idx_acknowledge_id UNIQUE (acknowledge_id))
 
  select.expired.sql = SELECT * FROM retry_store \
  WHERE (acknowledged='F' AND \
- (retries_to_date >= total_retries AND total_retries != -1))
+ (retries_to_date >= total_retries AND total_retries != -1))
* *

@@ -97,14 +95,15 @@ CONSTRAINT idx_acknowledge_id UNIQUE (acknowledge_id)) * The create.sql script is always executed upon initialisation, * any errors are discarded *

+ *

+ * Partial implementation of behaviour common to retry services. + *

*/ -@XStreamAlias("retry-store-jdbc") -@ComponentProfile(summary = "Store message for retry in a database using jdbc", since = "4.9.0") -@DisplayOrder(order = { "sqlPropertiesFile" }) -@Slf4j -public class JdbcRetryStore implements RetryStore { +@NoArgsConstructor +public abstract class JdbcRetryServiceImp extends JdbcService { + private static final String RETRY_STORE_PROPERTIES = "retry-store-derby.properties"; private static final String CREATE_SQL = "create.sql"; private static final String INSERT_SQL = "insert.sql"; @@ -114,41 +113,41 @@ public class JdbcRetryStore implements RetryStore { private static final String UPDATE_RETRY_SQL = "update-retry.sql"; private static final String DELETE_ACKNOWLEGED_SQL = "delete.acknowleged.sql"; private static final String ACKNOWLEDGE_SQL = "acknowledge.sql"; - + private transient Properties sqlStatements; private transient MimeEncoder encoder; - private AdaptrisConnection connection; - private Connection sqlConnection; + + protected static AdaptrisMarshaller marshaller; + + static { // only create marshaller once + marshaller = DefaultMarshaller.getDefaultMarshaller(); + } + /** * Set the sql properties file to use, by default it will look for the value * "retry-store-derby.properties" */ + @Getter @Setter @NotBlank @InputFieldDefault(value = RETRY_STORE_PROPERTIES) private String sqlPropertiesFile; + + @InputFieldDefault(value = "true") + private boolean pruneAcknowledged; - /** - *

- * Creates a new instance. Default properties file is - * retry-store-derby.properties. - *

- */ - public JdbcRetryStore() { + /** @see com.adaptris.core.AdaptrisComponent#init() */ + @Override + protected void initJdbcService() throws CoreException { encoder = new MimeEncoder(); encoder.setRetainUniqueId(true); - setSqlPropertiesFile(RETRY_STORE_PROPERTIES); - } - - @Override - public void prepare() throws CoreException { - Args.notBlank(getSqlPropertiesFile(), "sqlPropertiesFile"); - } - - @Override - public void init() throws CoreException { + if (getConnection() == null) { + throw new CoreException("DatabaseConnection is null in service"); + } + getConnection().init(); + if (sqlStatements == null) { sqlStatements = new Properties(); @@ -164,23 +163,80 @@ public void init() throws CoreException { } catch (Exception e) { throw new CoreException("problem loading file [" + getSqlPropertiesFile() + "]"); } - } - - if (sqlConnection == null) { try { - sqlConnection = getConnection(); + createStoreTable(); + log.debug("store table created"); } catch (Exception e) { - throw new CoreException("error connecting to the Database"); } } + } + + /** @see com.adaptris.core.ServiceImp#start() */ + @Override + public void start() throws CoreException { + getConnection().start(); + } + + @Override + protected void prepareService() throws CoreException { + // TODO Auto-generated method stub + + } + + @Override + protected void startService() throws CoreException { + // TODO Auto-generated method stub + + } + + /** @see com.adaptris.core.AdaptrisComponent#close() */ + @Override + protected void closeJdbcService() { + getConnection().close(); + } + + public final void doService(AdaptrisMessage msg) throws ServiceException { + pruneAcknowledged(); + performService(msg); + } + + protected abstract void performService(AdaptrisMessage msg) + throws ServiceException; + + private void pruneAcknowledged() { try { - createStoreTable(); - log.debug("store table created"); - } catch (Exception e) { + if (isPruneAcknowledged()) { + log.debug("Pruning Previously Acknowledged Messages"); + deleteAcknowledged(); + } + } + catch (Exception e) { + log.warn("Ignoring exception while pruning acknowledged messages[" + + e.getMessage() + "]"); } } - @Override + /** + * @return the pruneAcknowledged + */ + public boolean getPruneAcknowledged() { + return pruneAcknowledged; + } + + /** + * Specify whether to delete messages from the underlying store if they have + * already been acknowledged. + * + * @param b the pruneAcknowledged to set + */ + public void setPruneAcknowledged(boolean b) { + this.pruneAcknowledged = b; + } + + private boolean isPruneAcknowledged() { + return BooleanUtils.toBooleanDefaultIfNull(getPruneAcknowledged(), true); + } + public void write(AdaptrisMessage msg) throws InterlokException { PreparedStatement ps = null; validateMessage(msg); @@ -189,7 +245,7 @@ public void write(AdaptrisMessage msg) throws InterlokException { Integer.parseInt(msg.getMetadataValue(Constants.RETRIES_KEY)), Integer.valueOf(0), msg.getMetadataValue(Constants.MARSHALLED_SERVICE_KEY).getBytes(), "F" }; try { - ps = prepareStatementWithParameters(sqlConnection, sqlStatements.getProperty(INSERT_SQL), params); + ps = prepareStatementWithParameters(sqlStatements.getProperty(INSERT_SQL), params); log.trace("executing insert statement"); ps.executeUpdate(); } catch (SQLException e) { @@ -199,12 +255,11 @@ public void write(AdaptrisMessage msg) throws InterlokException { } } - @Override public boolean delete(String msgId) throws InterlokException { PreparedStatement ps = null; Object[] params = new Object[] { msgId }; try { - ps = prepareStatementWithParameters(sqlConnection, sqlStatements.getProperty(DELETE_SQL), params); + ps = prepareStatementWithParameters(sqlStatements.getProperty(DELETE_SQL), params); log.trace("executing delete statement"); ps.executeUpdate(); return true; @@ -215,29 +270,11 @@ public boolean delete(String msgId) throws InterlokException { } } - @Override - public Iterable report() throws InterlokException { - return Collections.EMPTY_LIST; - } - - @Override - public AdaptrisMessage buildForRetry(String msgId, Map metadata, AdaptrisMessageFactory factory) - throws InterlokException { - return null; - } - - @Override - public Map getMetadata(String msgId) throws InterlokException { - // TODO Auto-generated method stub - return null; - } - - @Override public void acknowledge(String acknowledgeId) throws InterlokException { PreparedStatement ps = null; Object[] params = { Constants.ACKNOWLEDGED, new Date(), acknowledgeId }; try { - ps = prepareStatementWithParameters(sqlConnection, sqlStatements.getProperty(ACKNOWLEDGE_SQL), params); + ps = prepareStatementWithParameters(sqlStatements.getProperty(ACKNOWLEDGE_SQL), params); log.trace("executing update statement"); ps.executeUpdate(); } catch (SQLException e) { @@ -247,11 +284,10 @@ public void acknowledge(String acknowledgeId) throws InterlokException { } } - @Override public void deleteAcknowledged() throws InterlokException { PreparedStatement ps = null; try { - ps = prepareStatementWithoutParameters(sqlConnection, sqlStatements.getProperty(DELETE_ACKNOWLEGED_SQL)); + ps = prepareStatementWithoutParameters(sqlStatements.getProperty(DELETE_ACKNOWLEGED_SQL)); log.trace("executing delete statement"); ps.executeUpdate(); } catch (SQLException e) { @@ -261,20 +297,19 @@ public void deleteAcknowledged() throws InterlokException { } } - @Override public List obtainExpiredMessages() throws InterlokException { List result = new ArrayList(); PreparedStatement ps = null; ResultSet rs = null; try { - ps = prepareStatementWithoutParameters(sqlConnection, sqlStatements.getProperty(SELECT_EXPIRED_SQL)); + ps = prepareStatementWithoutParameters(sqlStatements.getProperty(SELECT_EXPIRED_SQL)); log.trace("executing select statement"); rs = ps.executeQuery(); if (rs == null) { return result; } while (rs.next()) { - JdbcRetryStoreEntry resultRow = mapRow(rs); + JdbcRetryRowEntry resultRow = mapRow(rs); result.add(convert(resultRow)); } } catch (SQLException e) { @@ -287,20 +322,19 @@ public List obtainExpiredMessages() throws InterlokException { return result; } - @Override public List obtainMessagesToRetry() throws InterlokException { PreparedStatement ps = null; ResultSet rs = null; List result = new ArrayList(); try { - ps = prepareStatementWithoutParameters(sqlConnection, sqlStatements.getProperty(RETRY_SQL)); + ps = prepareStatementWithoutParameters(sqlStatements.getProperty(RETRY_SQL)); log.trace("executing select statement"); rs = ps.executeQuery(); if (rs == null) { return result; } while (rs.next()) { - JdbcRetryStoreEntry resultRow = mapRow(rs); + JdbcRetryRowEntry resultRow = mapRow(rs); long now = System.currentTimeMillis(); long lastRetry = resultRow.getUpdatedOn().getTime(); int interval = resultRow.getRetryInterval(); @@ -318,12 +352,11 @@ public List obtainMessagesToRetry() throws InterlokException { return result; } - @Override public void updateRetryCount(String messageId) throws InterlokException { PreparedStatement ps = null; Object[] params = { new Date(), messageId }; try { - ps = prepareStatementWithParameters(sqlConnection, sqlStatements.getProperty(UPDATE_RETRY_SQL), params); + ps = prepareStatementWithParameters(sqlStatements.getProperty(UPDATE_RETRY_SQL), params); log.trace("executing update statement"); ps.executeUpdate(); } catch (SQLException e) { @@ -362,7 +395,7 @@ private void validateMessage(AdaptrisMessage msg) throws InterlokException { } } - private AdaptrisMessage convert(JdbcRetryStoreEntry retryStoreEntry) throws InterlokException { + private AdaptrisMessage convert(JdbcRetryRowEntry retryStoreEntry) throws InterlokException { AdaptrisMessage result = encoder.decode(retryStoreEntry.getEncodedMessage()); @@ -379,7 +412,7 @@ private AdaptrisMessage convert(JdbcRetryStoreEntry retryStoreEntry) throws Inte private void createStoreTable() throws Exception { PreparedStatement ps = null; try { - ps = prepareStatementWithoutParameters(sqlConnection, sqlStatements.getProperty(CREATE_SQL)); + ps = prepareStatementWithoutParameters(sqlStatements.getProperty(CREATE_SQL)); log.trace("Executing create statement"); ps.executeUpdate(); } catch (SQLException e) { @@ -389,9 +422,10 @@ private void createStoreTable() throws Exception { } } - private static PreparedStatement prepareStatementWithParameters(Connection c, String query, Object[] parameters) - throws SQLException { - PreparedStatement preparedStatement = c.prepareStatement(query); + private PreparedStatement prepareStatementWithParameters(String query, Object[] parameters) throws SQLException { + Connection conn; + conn = makeConnection(); + PreparedStatement preparedStatement = conn.prepareStatement(query); int count = 1; for (Object o : parameters) { preparedStatement.setObject(count, o); @@ -400,12 +434,14 @@ private static PreparedStatement prepareStatementWithParameters(Connection c, St return preparedStatement; } - private static PreparedStatement prepareStatementWithoutParameters(Connection c, String query) throws SQLException { - return c.prepareStatement(query); + private PreparedStatement prepareStatementWithoutParameters(String query) throws SQLException { + Connection conn; + conn = makeConnection(); + return conn.prepareStatement(query); } - private JdbcRetryStoreEntry mapRow(ResultSet rs) throws SQLException { - JdbcRetryStoreEntry result = new JdbcRetryStoreEntry(); + private JdbcRetryRowEntry mapRow(ResultSet rs) throws SQLException { + JdbcRetryRowEntry result = new JdbcRetryRowEntry(); result.setMessageId(rs.getString("message_id")); result.setAcknowledgeId(rs.getString("acknowledge_id")); @@ -427,16 +463,9 @@ private JdbcRetryStoreEntry mapRow(ResultSet rs) throws SQLException { return result; } - public void setConnection(AdaptrisConnection connection) { - this.connection = connection; + private Connection makeConnection() throws SQLException { + return getConnection().retrieveConnection(DatabaseConnection.class).connect(); } - public Connection getConnection() throws SQLException { - return this.connection.retrieveConnection(DatabaseConnection.class).connect(); - } - @Override - public void makeConnection(AdaptrisConnection connection) { - setConnection(connection); - } } diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/StoreMessageForRetryService.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryService.java similarity index 92% rename from interlok-core/src/main/java/com/adaptris/core/jdbc/retry/StoreMessageForRetryService.java rename to interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryService.java index 43b4d4093..7cffbed26 100644 --- a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/StoreMessageForRetryService.java +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryService.java @@ -1,4 +1,4 @@ -package com.adaptris.core.jdbc.retry; +package com.adaptris.core.services.jdbc.retry; import javax.validation.Valid; import javax.validation.constraints.NotNull; @@ -14,6 +14,7 @@ import com.adaptris.core.NullService; import com.adaptris.core.Service; import com.adaptris.core.ServiceException; +import com.adaptris.core.jdbc.retry.Constants; import com.adaptris.interlok.InterlokException; import com.thoughtworks.xstream.annotations.XStreamAlias; @@ -23,7 +24,7 @@ *

*

* This class supports both synchronous and asynchronous acknowledgement. This is controlled by the field - * {@linkplain StoreMessageForRetryServiceTest#setAsynchronousAcknowledgment(boolean)}. + * {@linkplain JdbcStoreMessageForRetryService#setAsynchronousAcknowledgment(boolean)}. *

*

* A message is deemed to be synchronously acknowledged if the wrapped service completes normally. In such cases it is not added to @@ -48,12 +49,12 @@ *

*/ -@XStreamAlias("store-message-for-retry-service") +@XStreamAlias("jdbc-store-message-for-retry-service") @AdapterComponent -@ComponentProfile(summary = "Wraps an interlok service and gives the option to store the message in a retry store in the event of an exception.", - since = "4.9.0", tag = "retry") -@DisplayOrder(order = {"asynchronousAcknowledgment", "asyncAutoRetryOnFail", "retryStore"}) -public class StoreMessageForRetryService extends RetryServiceImp { +@ComponentProfile(summary = "Wraps an interlok service and gives the option to store the message into a Database table in the event of an exception.", + since = "5.0.0", tag = "retry") +@DisplayOrder(order = {"asynchronousAcknowledgment", "asyncAutoRetryOnFail", "jdbc, retry"}) +public class JdbcStoreMessageForRetryService extends JdbcRetryServiceImp { // persistent @NotNull @@ -74,7 +75,7 @@ public class StoreMessageForRetryService extends RetryServiceImp { *
  • asynchronousAcknowledgment is false
    • *
  • asyncAutoRetryOnFail is true

    • */ - public StoreMessageForRetryService() { + public JdbcStoreMessageForRetryService() { super(); setService(new NullService()); setAsyncAutoRetryOnFail(true); @@ -97,7 +98,7 @@ public void stop() { /** * - * @see RetryServiceImpTest#performService(com.adaptris.core.AdaptrisMessage) + * @see JdbcRetryServiceImp#performService(com.adaptris.core.AdaptrisMessage) */ @Override protected void performService(AdaptrisMessage msg) throws ServiceException { @@ -163,7 +164,7 @@ private String applyMetadata(AdaptrisMessage msg) throws ServiceException { private void insertMessageForRetry(AdaptrisMessage msg) throws ServiceException { try { String ackId = applyMetadata(msg); - getRetryStore().write(msg); + write(msg); log.debug("Storing [" + ackId + "] for future acknowledgement"); } diff --git a/interlok-core/src/test/java/com/adaptris/core/jdbc/retry/JdbcRetryStoreTest.java b/interlok-core/src/test/java/com/adaptris/core/jdbc/retry/JdbcRetryStoreTest.java deleted file mode 100644 index 7a1585624..000000000 --- a/interlok-core/src/test/java/com/adaptris/core/jdbc/retry/JdbcRetryStoreTest.java +++ /dev/null @@ -1,28 +0,0 @@ -package com.adaptris.core.jdbc.retry; - -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; - -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.TestInstance; -import org.junit.jupiter.api.TestInstance.Lifecycle; - -import com.adaptris.interlok.junit.scaffolding.BaseCase; - -import java.sql.Connection; -import java.sql.DriverManager; - -@TestInstance(Lifecycle.PER_CLASS) -public class JdbcRetryStoreTest extends BaseCase { - - protected static final String JDBC_RETRY_STORE_DRIVER = "jdbc.retrystore.driver"; - protected static final String JDBC_RETRY_STORE_URL = "jdbc.retrystore.url"; - protected static final String JDBC_RETRY_STORE_PROPERTIES = "jdbc.RetryStore.properties.file.destination"; - - private JdbcRetryStore jdbcRetryStore; - private Connection sqlConnection; - -} \ No newline at end of file diff --git a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeServiceTest.java b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeServiceTest.java new file mode 100644 index 000000000..b8c791987 --- /dev/null +++ b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeServiceTest.java @@ -0,0 +1,14 @@ +package com.adaptris.core.services.jdbc.retry; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + + +public class JdbcAcknowledgeServiceTest { + +} \ No newline at end of file diff --git a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesServiceTest.java b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesServiceTest.java new file mode 100644 index 000000000..d8bc5d15f --- /dev/null +++ b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesServiceTest.java @@ -0,0 +1,14 @@ +package com.adaptris.core.services.jdbc.retry; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + + +public class JdbcRetryMessagesServiceTest { + +} \ No newline at end of file diff --git a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryServiceTest.java b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryServiceTest.java new file mode 100644 index 000000000..b8ab3e174 --- /dev/null +++ b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryServiceTest.java @@ -0,0 +1,14 @@ +package com.adaptris.core.services.jdbc.retry; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; + + +public class JdbcStoreMessageForRetryServiceTest { + +} \ No newline at end of file From 3956438f2fd276fbbb41d2c132d9dbe15cd23d6c Mon Sep 17 00:00:00 2001 From: James Wickham Date: Wed, 27 Sep 2023 10:50:38 +0100 Subject: [PATCH 02/10] changing access modifiers to public on getters and setters and constants so the required classes can access it still now they have been moved to another package. --- .../adaptris/core/jdbc/retry/Constants.java | 4 +- .../core/jdbc/retry/JdbcRetryRowEntry.java | 42 +++++++++---------- 2 files changed, 23 insertions(+), 23 deletions(-) diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/Constants.java b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/Constants.java index 35a91f40c..cebe8e540 100644 --- a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/Constants.java +++ b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/Constants.java @@ -58,8 +58,8 @@ public final class Constants { */ public static final String ASYNC_AUTO_RETRY = "retryAsyncRetry"; - protected static final String ACKNOWLEDGED = "T"; - protected static final String NOT_ACKNOWLEDGED = "F"; + public static final String ACKNOWLEDGED = "T"; + public static final String NOT_ACKNOWLEDGED = "F"; private Constants() { // no instances diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java index 1a68d43f9..6ec4caa3f 100644 --- a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java +++ b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java @@ -8,7 +8,7 @@ * 'retry store database'. *

    */ -class JdbcRetryStoreEntry { +public class JdbcRetryRowEntry { private String messageId; private String acknowledgeId; @@ -21,83 +21,83 @@ class JdbcRetryStoreEntry { private Date insertedOn; private Date updatedOn; - Date getInsertedOn() { + public Date getInsertedOn() { return insertedOn; } - void setInsertedOn(Date d) { + public void setInsertedOn(Date d) { this.insertedOn = d; } - Date getUpdatedOn() { + public Date getUpdatedOn() { return updatedOn; } - void setUpdatedOn(Date d) { + public void setUpdatedOn(Date d) { this.updatedOn = d; } - String getAcknowledgeId() { + public String getAcknowledgeId() { return acknowledgeId; } - void setAcknowledgeId(String s) { + public void setAcknowledgeId(String s) { this.acknowledgeId = s; } - byte[] getEncodedMessage() { + public byte[] getEncodedMessage() { return encodedMessage; } - void setEncodedMessage(byte[] b) { + public void setEncodedMessage(byte[] b) { this.encodedMessage = b; } - String getMessageId() { + public String getMessageId() { return messageId; } - void setMessageId(String s) { + public void setMessageId(String s) { this.messageId = s; } - int getTotalRetries() { + public int getTotalRetries() { return totalRetries; } - void setTotalRetries(int i) { + public void setTotalRetries(int i) { this.totalRetries = i; } - int getRetriesToDate() { + public int getRetriesToDate() { return retriesToDate; } - void setRetriesToDate(int i) { + public void setRetriesToDate(int i) { this.retriesToDate = i; } - int getRetryInterval() { + public int getRetryInterval() { return retryInterval; } - void setRetryInterval(int i) { + public void setRetryInterval(int i) { this.retryInterval = i; } - String getMarshalledService() { + public String getMarshalledService() { return marshalledService; } - void setMarshalledService(String s) { + public void setMarshalledService(String s) { this.marshalledService = s; } - boolean getAcknowledged() { + public boolean getAcknowledged() { return acknowledged; } - void setAcknowledged(boolean b) { + public void setAcknowledged(boolean b) { this.acknowledged = b; } } From 2a3fb2199fb5d2962610b3d84ff7259b402dedc0 Mon Sep 17 00:00:00 2001 From: James Wickham Date: Wed, 27 Sep 2023 10:51:37 +0100 Subject: [PATCH 03/10] removing unimplemented methods from the RetryStore interface and classes that implement it. Services have now been split out so no longer need to use the same interface. --- .../jetty/retry/FilesystemRetryStore.java | 30 --------- .../core/http/jetty/retry/RetryStore.java | 67 ------------------- .../http/jetty/retry/InMemoryRetryStore.java | 32 --------- .../http/jetty/retry/RetryFromJettyTest.java | 31 --------- .../core/http/jetty/retry/RetryStoreTest.java | 33 --------- 5 files changed, 193 deletions(-) diff --git a/interlok-core/src/main/java/com/adaptris/core/http/jetty/retry/FilesystemRetryStore.java b/interlok-core/src/main/java/com/adaptris/core/http/jetty/retry/FilesystemRetryStore.java index 3cafd16fd..ea21f08a8 100644 --- a/interlok-core/src/main/java/com/adaptris/core/http/jetty/retry/FilesystemRetryStore.java +++ b/interlok-core/src/main/java/com/adaptris/core/http/jetty/retry/FilesystemRetryStore.java @@ -21,7 +21,6 @@ import org.apache.commons.lang3.BooleanUtils; import com.adaptris.annotation.ComponentProfile; import com.adaptris.annotation.DisplayOrder; -import com.adaptris.core.AdaptrisConnection; import com.adaptris.core.AdaptrisMessage; import com.adaptris.core.AdaptrisMessageFactory; import com.adaptris.core.CoreException; @@ -231,33 +230,4 @@ public FilesystemRetryStore withBaseUrl(String s) { return this; } - @Override - public void acknowledge(String acknowledgeId) throws InterlokException { - // null implementation - } - - @Override - public void deleteAcknowledged() throws InterlokException { - // null implementation - } - - @Override - public List obtainExpiredMessages() throws InterlokException { - return null; // null implementation - } - - @Override - public List obtainMessagesToRetry() throws InterlokException { - return null; // null implementation - } - - @Override - public void updateRetryCount(String messageId) throws InterlokException { - // null implementation - } - - @Override - public void makeConnection(AdaptrisConnection connection) { - // null implementation - } } diff --git a/interlok-core/src/main/java/com/adaptris/core/http/jetty/retry/RetryStore.java b/interlok-core/src/main/java/com/adaptris/core/http/jetty/retry/RetryStore.java index a4e9d9d67..a5fb533a6 100644 --- a/interlok-core/src/main/java/com/adaptris/core/http/jetty/retry/RetryStore.java +++ b/interlok-core/src/main/java/com/adaptris/core/http/jetty/retry/RetryStore.java @@ -1,10 +1,8 @@ package com.adaptris.core.http.jetty.retry; import java.util.Collections; -import java.util.List; import java.util.Map; -import com.adaptris.core.AdaptrisConnection; import com.adaptris.core.AdaptrisMessage; import com.adaptris.core.AdaptrisMessageFactory; import com.adaptris.core.ComponentLifecycle; @@ -91,69 +89,4 @@ default boolean delete(String msgId) throws InterlokException { default void prepare() throws CoreException { } - /** - *

    - * Acknowledge that the message with the passed ID has now been successfully - * processed and should not be retried again. NB this method does not throw an - * Exception if the acknowledge ID does not exist in the store. - *

    - * - * @param acknowledgeId the acknowledge ID of the message to acknowledge - * @throws InterlokException wrapping any Exception which occurs - */ - void acknowledge(String acknowledgeId) throws InterlokException; - - /** - * Delete any messages that have been successfully Acknowledged. - * - * @throws InterlokException wrapping any Exception which occurs - */ - void deleteAcknowledged() throws InterlokException; - - /** - *

    - * Obtain a list of AdaptrisMessages which meet the expiration - * criteria. - * In the most abstract sense, expired messages are those that have exceeded - * their max retry count but not yet been acknowledged. - *

    - * - * @return a list of AdaptrisMessages which meet the expiration - * criteria. - * @throws InterlokException wrapping any Exception which occurs - */ - List obtainExpiredMessages() throws InterlokException; - - /** - *

    - * Obtain a list of AdaptrisMessages which meet the criteria - * for retrying. - *

    - * - * @return a list of AdaptrisMessages which meet the criteria - * for retrying - * @throws InterlokException wrapping any Exception which occurs - */ - List obtainMessagesToRetry() throws InterlokException; - - /** - *

    - * Update the number of retries which have taken place for the message with - * the passed ID. NB this method does not throw an Exception if an attempt is - * made to update the retry count for a message ID which does not exist in the - * store. - *

    - * - * @param messageId the ID of the message to update - * @throws InterlokException wrapping any Exception which occurs - */ - void updateRetryCount(String messageId) throws InterlokException; - - /** - *

    - * Used for any implementations that have a connected RetryStore - *

    - */ - void makeConnection(AdaptrisConnection connection); - } \ No newline at end of file diff --git a/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/InMemoryRetryStore.java b/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/InMemoryRetryStore.java index ebe31ee10..30e31cc00 100644 --- a/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/InMemoryRetryStore.java +++ b/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/InMemoryRetryStore.java @@ -2,11 +2,9 @@ import java.util.Collections; import java.util.HashMap; -import java.util.List; import java.util.Map; import java.util.stream.Collectors; -import com.adaptris.core.AdaptrisConnection; import com.adaptris.core.AdaptrisMessage; import com.adaptris.core.AdaptrisMessageFactory; import com.adaptris.interlok.InterlokException; @@ -58,34 +56,4 @@ public Iterable report() throws InterlokException { public static void removeAll() { STORE.clear(); } - - @Override - public void acknowledge(String acknowledgeId) throws InterlokException { - // null implementation - } - - @Override - public void deleteAcknowledged() throws InterlokException { - // null implementation - } - - @Override - public List obtainExpiredMessages() throws InterlokException { - return null; // null implementation - } - - @Override - public List obtainMessagesToRetry() throws InterlokException { - return null; // null implementation - } - - @Override - public void updateRetryCount(String messageId) throws InterlokException { - // null implementation - } - - @Override - public void makeConnection(AdaptrisConnection connection) { - // null implementation - } } diff --git a/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/RetryFromJettyTest.java b/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/RetryFromJettyTest.java index 600ac06f2..1497b831f 100644 --- a/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/RetryFromJettyTest.java +++ b/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/RetryFromJettyTest.java @@ -19,7 +19,6 @@ import org.junit.jupiter.api.Test; import com.adaptris.core.Adapter; -import com.adaptris.core.AdaptrisConnection; import com.adaptris.core.AdaptrisMessage; import com.adaptris.core.AdaptrisMessageFactory; import com.adaptris.core.ChannelList; @@ -412,35 +411,5 @@ public Iterable report() throws InterlokException { throw new UnsupportedOperationException(); } - @Override - public void acknowledge(String acknowledgeId) throws InterlokException { - // null implementation - } - - @Override - public void deleteAcknowledged() throws InterlokException { - // null implementation - } - - @Override - public List obtainExpiredMessages() throws InterlokException { - return null; // null implementation - } - - @Override - public List obtainMessagesToRetry() throws InterlokException { - return null; // null implementation - } - - @Override - public void updateRetryCount(String messageId) throws InterlokException { - // null implementation - } - - @Override - public void makeConnection(AdaptrisConnection connection) { - // null implementation - } } - } diff --git a/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/RetryStoreTest.java b/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/RetryStoreTest.java index 1477132e2..167368531 100644 --- a/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/RetryStoreTest.java +++ b/interlok-core/src/test/java/com/adaptris/core/http/jetty/retry/RetryStoreTest.java @@ -4,7 +4,6 @@ import static org.junit.jupiter.api.Assertions.assertNull; import java.util.Collections; -import java.util.List; import java.util.Map; import org.junit.jupiter.api.AfterEach; @@ -12,7 +11,6 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import com.adaptris.core.AdaptrisConnection; import com.adaptris.core.AdaptrisMessage; import com.adaptris.core.AdaptrisMessageFactory; import com.adaptris.core.util.LifecycleHelper; @@ -64,35 +62,4 @@ public Map getMetadata(String msgId) throws InterlokException { return Collections.EMPTY_MAP; } - @Override - public void acknowledge(String acknowledgeId) throws InterlokException { - // null implementation - } - - @Override - public void deleteAcknowledged() throws InterlokException { - // null implementation - } - - @Override - public List obtainExpiredMessages() throws InterlokException { - return null; // null implementation - } - - @Override - public List obtainMessagesToRetry() throws InterlokException { - return null; // null implementation - } - - - @Override - public void updateRetryCount(String messageId) throws InterlokException { - // null implementation - } - - @Override - public void makeConnection(AdaptrisConnection connection) { - // null implementation - } - } From 12e9fb99fdd25449700e0523de26905817bf62a2 Mon Sep 17 00:00:00 2001 From: James Wickham Date: Wed, 27 Sep 2023 16:49:55 +0100 Subject: [PATCH 04/10] removing old properties file and replacing with newly named one --- ...try-store.properties => retry-store-derby.properties} | 2 ++ .../src/test/resources/unit-tests.properties.template | 9 +++------ 2 files changed, 5 insertions(+), 6 deletions(-) rename interlok-core/src/test/resources/{retry-store.properties => retry-store-derby.properties} (97%) diff --git a/interlok-core/src/test/resources/retry-store.properties b/interlok-core/src/test/resources/retry-store-derby.properties similarity index 97% rename from interlok-core/src/test/resources/retry-store.properties rename to interlok-core/src/test/resources/retry-store-derby.properties index a8fba73ec..a261387de 100644 --- a/interlok-core/src/test/resources/retry-store.properties +++ b/interlok-core/src/test/resources/retry-store-derby.properties @@ -1,3 +1,5 @@ +drop.sql = DROP TABLE retry_store + create.sql = CREATE TABLE retry_store (message_id VARCHAR(256) NOT NULL, acknowledge_id VARCHAR(256) NOT NULL, message BLOB NOT NULL, retry_interval INTEGER NOT NULL, total_retries INTEGER NOT NULL, retries_to_date INTEGER NOT NULL, marshalled_service BLOB NOT NULL, acknowledged CHAR NOT NULL, updated_on TIMESTAMP, inserted_on TIMESTAMP, CONSTRAINT pk_message_id PRIMARY KEY (message_id), CONSTRAINT idx_acknowledge_id UNIQUE (acknowledge_id)) insert.sql = INSERT INTO retry_store (message_id, acknowledge_id, message, retry_interval, total_retries, retries_to_date, marshalled_service, acknowledged, inserted_on, updated_on) VALUES (?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) diff --git a/interlok-core/src/test/resources/unit-tests.properties.template b/interlok-core/src/test/resources/unit-tests.properties.template index e69dd74a0..3a0cdec80 100644 --- a/interlok-core/src/test/resources/unit-tests.properties.template +++ b/interlok-core/src/test/resources/unit-tests.properties.template @@ -161,9 +161,9 @@ junit.jdbc.queryservice.driver=org.apache.derby.jdbc.EmbeddedDriver junit.jdbc.captureservice.url=jdbc:derby:memory:JdbcDataQueryService;create=true junit.jdbc.captureservice.driver=org.apache.derby.jdbc.EmbeddedDriver -junit.jdbc.retrystore.url=jdbc:derby:memory:retryStore;create=true -junit.jdbc.retrystore.driver=org.apache.derby.jdbc.EmbeddedDriver - +junit.jdbc.retryservice.url=jdbc:derby:memory:JdbcRetryService;create=true +junit.jdbc.retryservice.driver=org.apache.derby.jdbc.EmbeddedDriver +junit.jdbc.retryservice.sql.properties.file=file:///@BUILD_DIR@/src/test/resources/retry-store-derby.properties junit.fs.SizeBasedFilter=@BASE_DIR@/src/test/resources/readme.txt @@ -230,6 +230,3 @@ junit.urlhelper.local=file://localhost/@BASE_DIR@/src/test/resources/xstream-sta # junit.urlhelper.remote=https://development.adaptris.net/index.html junit.urlhelper.remote=https://raw.githubusercontent.com/adaptris-labs/interlok-build-parent/master/build.gradle junit.urlhelper.classpath=xstream-standard.xml - -junit.retry.baseUrl=file://localhost/@BUILD_DIR@/tmp/retry-store -junit.jdbc.RetryStore.properties.file.destination=file:///@BUILD_DIR@/src/test/resources/retry-store.properties From d5aa71287df0f497911203a0cd2cea95c3e07b98 Mon Sep 17 00:00:00 2001 From: James Wickham Date: Thu, 28 Sep 2023 12:59:54 +0100 Subject: [PATCH 05/10] correcting retry-store properties file path and adding back a uni test property removed by mistake. --- .../src/test/resources/unit-tests.properties.template | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/interlok-core/src/test/resources/unit-tests.properties.template b/interlok-core/src/test/resources/unit-tests.properties.template index 3a0cdec80..6b0c295fe 100644 --- a/interlok-core/src/test/resources/unit-tests.properties.template +++ b/interlok-core/src/test/resources/unit-tests.properties.template @@ -163,7 +163,7 @@ junit.jdbc.captureservice.driver=org.apache.derby.jdbc.EmbeddedDriver junit.jdbc.retryservice.url=jdbc:derby:memory:JdbcRetryService;create=true junit.jdbc.retryservice.driver=org.apache.derby.jdbc.EmbeddedDriver -junit.jdbc.retryservice.sql.properties.file=file:///@BUILD_DIR@/src/test/resources/retry-store-derby.properties +junit.jdbc.retryservice.sql.properties.file=@BUILD_DIR@/resources/test/retry-store-derby.properties junit.fs.SizeBasedFilter=@BASE_DIR@/src/test/resources/readme.txt @@ -230,3 +230,4 @@ junit.urlhelper.local=file://localhost/@BASE_DIR@/src/test/resources/xstream-sta # junit.urlhelper.remote=https://development.adaptris.net/index.html junit.urlhelper.remote=https://raw.githubusercontent.com/adaptris-labs/interlok-build-parent/master/build.gradle junit.urlhelper.classpath=xstream-standard.xml +junit.retry.baseUrl=file://localhost/@BUILD_DIR@/tmp/retry-store \ No newline at end of file From 95c8d539e16d23ca2dc871ae83b4fa4f2f6f360a Mon Sep 17 00:00:00 2001 From: James Wickham Date: Fri, 29 Sep 2023 18:01:04 +0100 Subject: [PATCH 06/10] not part of this change but it kept flagging as an error during javadoc generation so I've fixed it. --- .../adaptris/core/http/client/net/HttpRequestServiceImpl.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/interlok-core/src/main/java/com/adaptris/core/http/client/net/HttpRequestServiceImpl.java b/interlok-core/src/main/java/com/adaptris/core/http/client/net/HttpRequestServiceImpl.java index a18406115..b572699c1 100644 --- a/interlok-core/src/main/java/com/adaptris/core/http/client/net/HttpRequestServiceImpl.java +++ b/interlok-core/src/main/java/com/adaptris/core/http/client/net/HttpRequestServiceImpl.java @@ -102,7 +102,7 @@ public abstract class HttpRequestServiceImpl extends ServiceImp { * Set the read timeout. *

    * Note that any read timeout will be overridden by the timeout value passed in via the - * {{@link #request(AdaptrisMessage, long)} method; if it is not the same as + * {{@link com.adaptris.core.RequestReplyProducerImp#request(AdaptrisMessage, long)} method; if it is not the same as * {@value com.adaptris.core.http.HttpConstants#DEFAULT_SOCKET_TIMEOUT} *

    */ From 049d398a352bc78a629d029e6d3b0cfbdb3b2982 Mon Sep 17 00:00:00 2001 From: James Wickham Date: Fri, 29 Sep 2023 18:01:20 +0100 Subject: [PATCH 07/10] correction to javadoc wording. --- .../java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java index 6ec4caa3f..0145ed416 100644 --- a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java +++ b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/JdbcRetryRowEntry.java @@ -4,8 +4,8 @@ /** *

    - * Represents an entry in retry_store, the sole table in the - * 'retry store database'. + * Represents an entry in retry table, the sole table in the + * 'retry database'. *

    */ public class JdbcRetryRowEntry { From ac67a8a9fe6ef5ffe797b122ee7a851d93c036cc Mon Sep 17 00:00:00 2001 From: James Wickham Date: Fri, 29 Sep 2023 18:01:38 +0100 Subject: [PATCH 08/10] adding package information for javadocs --- .../main/java/com/adaptris/core/jdbc/retry/package-info.java | 4 ++++ .../com/adaptris/core/services/jdbc/retry/package-info.java | 4 ++++ 2 files changed, 8 insertions(+) create mode 100644 interlok-core/src/main/java/com/adaptris/core/jdbc/retry/package-info.java create mode 100644 interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/package-info.java diff --git a/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/package-info.java b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/package-info.java new file mode 100644 index 000000000..5bcc6e3f8 --- /dev/null +++ b/interlok-core/src/main/java/com/adaptris/core/jdbc/retry/package-info.java @@ -0,0 +1,4 @@ +/** + * Supporting Classes for the {@link com.adaptris.core.services.jdbc.retry} service classes. + */ +package com.adaptris.core.jdbc.retry; diff --git a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/package-info.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/package-info.java new file mode 100644 index 000000000..bf2d602fe --- /dev/null +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/package-info.java @@ -0,0 +1,4 @@ +/** + * For storing, acknowledging and retrying messages using JDBC. + */ +package com.adaptris.core.services.jdbc.retry; From 6ff78424a69b42c02d56dd66e6e788986447246a Mon Sep 17 00:00:00 2001 From: James Wickham Date: Fri, 29 Sep 2023 18:03:59 +0100 Subject: [PATCH 09/10] updating access modifiers so they don't have unessisary visibility. correcting javadocs updating how we handle the service's state --- .../jdbc/retry/JdbcAcknowledgeService.java | 22 +++--- .../jdbc/retry/JdbcRetryMessagesService.java | 50 ++++++------- .../jdbc/retry/JdbcRetryServiceImp.java | 70 ++++++------------- .../JdbcStoreMessageForRetryService.java | 36 ++++------ 4 files changed, 69 insertions(+), 109 deletions(-) diff --git a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeService.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeService.java index 277c8ddb4..d670db1b9 100644 --- a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeService.java +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeService.java @@ -4,6 +4,7 @@ import com.adaptris.annotation.ComponentProfile; import com.adaptris.annotation.DisplayOrder; import com.adaptris.core.AdaptrisMessage; +import com.adaptris.core.CoreException; import com.adaptris.core.ServiceException; import com.adaptris.core.jdbc.retry.Constants; import com.adaptris.interlok.InterlokException; @@ -28,14 +29,21 @@ @ComponentProfile(summary = "processes asynchronous acknowledgements.", since = "5.0.0", tag = "jdbc, retry") @DisplayOrder(order = { "pruneExpired", "retryStore" }) public class JdbcAcknowledgeService extends JdbcRetryServiceImp { + + @Override + protected void startService() throws CoreException {} + + @Override + protected void stopService() {} /** - * - * @see JdbcRetryServiceImp#performService(com.adaptris.core.AdaptrisMessage) + * The main service method, which acknowledges a messages's entry in the retry database table. + * + * @see com.adaptris.core.Service#doService(com.adaptris.core.AdaptrisMessage) */ @Override - protected void performService(AdaptrisMessage msg) throws ServiceException { - + public void doService(AdaptrisMessage msg) throws ServiceException { + pruneAcknowledged(); String acknowledgeId = msg.getMetadataValue(Constants.ACKNOWLEDGE_ID_KEY); if (acknowledgeId == null) { log.debug(Constants.ACKNOWLEDGE_ID_KEY + " not available as metadata key or returned null"); @@ -48,9 +56,5 @@ protected void performService(AdaptrisMessage msg) throws ServiceException { throw new ServiceException(e); } } - - @Override - protected void stopService() { - // TODO Auto-generated method stub - } + } \ No newline at end of file diff --git a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesService.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesService.java index 0ad8ca5f3..4c48f3c8e 100644 --- a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesService.java +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesService.java @@ -17,6 +17,7 @@ import com.adaptris.core.ServiceException; import com.adaptris.core.StandaloneProducer; import com.adaptris.core.jdbc.retry.Constants; +import com.adaptris.core.util.LifecycleHelper; import com.adaptris.interlok.InterlokException; import com.thoughtworks.xstream.annotations.XStreamAlias; @@ -53,34 +54,27 @@ public JdbcRetryMessagesService() { setExpiredMessagesProducer(new StandaloneProducer()); setPruneExpired(false); } - - /** @see com.adaptris.core.ServiceImp#start() */ - @Override - public void start() throws CoreException { - super.start(); - getExpiredMessagesProducer().start(); - } - - /** @see com.adaptris.core.ServiceImp#stop() */ + + /** @see com.adaptris.core.AdaptrisComponent#start() */ @Override - public void stop() { - super.stop(); - getExpiredMessagesProducer().stop(); + protected void startService() throws CoreException { + LifecycleHelper.start(getExpiredMessagesProducer()); } - + + /** @see com.adaptris.core.AdaptrisComponent#stop() */ @Override protected void stopService() { - super.close(); - getExpiredMessagesProducer().close(); - getConnection().stop(); + LifecycleHelper.stop(getExpiredMessagesProducer()); } /** - * - * @see JdbcRetryServiceImp#performService(com.adaptris.core.AdaptrisMessage) + * The main service method, which retries an entry in the retry database table if it meets it's criteria. + * + * @see com.adaptris.core.Service#doService(com.adaptris.core.AdaptrisMessage) */ @Override - protected void performService(AdaptrisMessage msg) throws ServiceException { + public void doService(AdaptrisMessage msg) throws ServiceException { + pruneAcknowledged(); try { pruneExpired(); List retryMsgs = obtainMessagesToRetry(); @@ -109,8 +103,8 @@ private void doRetry(AdaptrisMessage retry) throws InterlokException { Service service = (Service) marshaller.unmarshal(marshalledService); - service.init(); - service.start(); + LifecycleHelper.init(service); + LifecycleHelper.start(service); try { log.debug("Retrying message [" + retry.getUniqueId() @@ -125,26 +119,22 @@ private void doRetry(AdaptrisMessage retry) throws InterlokException { } } finally { - service.stop(); - service.close(); + LifecycleHelper.stop(service); + LifecycleHelper.close(service); } } - private void handleSynchronous(AdaptrisMessage retry, Service service) - throws InterlokException { - + private void handleSynchronous(AdaptrisMessage retry, Service service) throws InterlokException { try { service.doService(retry); - acknowledge( - retry.getMetadataValue(Constants.ACKNOWLEDGE_ID_KEY)); + acknowledge(retry.getMetadataValue(Constants.ACKNOWLEDGE_ID_KEY)); } catch (InterlokException e) { updateRetryCount(retry.getUniqueId()); } } - private void handleAsynchronous(AdaptrisMessage retry, Service service) - throws InterlokException { + private void handleAsynchronous(AdaptrisMessage retry, Service service) throws InterlokException { try { service.doService(retry); updateRetryCount(retry.getUniqueId()); diff --git a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceImp.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceImp.java index 37310fb5b..7a3e46ce1 100644 --- a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceImp.java +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceImp.java @@ -21,7 +21,6 @@ import com.adaptris.core.CoreException; import com.adaptris.core.DefaultMarshaller; import com.adaptris.core.MimeEncoder; -import com.adaptris.core.ServiceException; import com.adaptris.core.jdbc.DatabaseConnection; import com.adaptris.core.jdbc.JdbcService; import com.adaptris.core.jdbc.retry.Constants; @@ -29,16 +28,12 @@ import com.adaptris.core.util.ExceptionHelper; import com.adaptris.core.util.JdbcUtil; import com.adaptris.interlok.InterlokException; -import com.adaptris.interlok.util.Args; import lombok.Getter; -import lombok.NoArgsConstructor; import lombok.Setter; /** - *

    - * JDBC services for storing and retrying messages with acknowledgment support. *

    *

    * If there is no explicit configuration of the sqlProperties then the SQL @@ -100,8 +95,6 @@ CONSTRAINT idx_acknowledge_id UNIQUE (acknowledge_id)) *

    */ - -@NoArgsConstructor public abstract class JdbcRetryServiceImp extends JdbcService { private static final String RETRY_STORE_PROPERTIES = "retry-store-derby.properties"; @@ -113,7 +106,7 @@ public abstract class JdbcRetryServiceImp extends JdbcService { private static final String UPDATE_RETRY_SQL = "update-retry.sql"; private static final String DELETE_ACKNOWLEGED_SQL = "delete.acknowleged.sql"; private static final String ACKNOWLEDGE_SQL = "acknowledge.sql"; - + private transient Properties sqlStatements; private transient MimeEncoder encoder; @@ -137,7 +130,12 @@ public abstract class JdbcRetryServiceImp extends JdbcService { @InputFieldDefault(value = "true") private boolean pruneAcknowledged; - + + public JdbcRetryServiceImp() { + setPruneAcknowledged(true); + setSqlPropertiesFile(RETRY_STORE_PROPERTIES); + } + /** @see com.adaptris.core.AdaptrisComponent#init() */ @Override protected void initJdbcService() throws CoreException { @@ -146,8 +144,6 @@ protected void initJdbcService() throws CoreException { if (getConnection() == null) { throw new CoreException("DatabaseConnection is null in service"); } - getConnection().init(); - if (sqlStatements == null) { sqlStatements = new Properties(); @@ -170,40 +166,17 @@ protected void initJdbcService() throws CoreException { } } } - - /** @see com.adaptris.core.ServiceImp#start() */ - @Override - public void start() throws CoreException { - getConnection().start(); - } - + + /** @see com.adaptris.core.AdaptrisComponent#prepare() */ @Override protected void prepareService() throws CoreException { - // TODO Auto-generated method stub - - } - - @Override - protected void startService() throws CoreException { - // TODO Auto-generated method stub - } /** @see com.adaptris.core.AdaptrisComponent#close() */ @Override - protected void closeJdbcService() { - getConnection().close(); - } + protected void closeJdbcService() {} - public final void doService(AdaptrisMessage msg) throws ServiceException { - pruneAcknowledged(); - performService(msg); - } - - protected abstract void performService(AdaptrisMessage msg) - throws ServiceException; - - private void pruneAcknowledged() { + void pruneAcknowledged() { try { if (isPruneAcknowledged()) { log.debug("Pruning Previously Acknowledged Messages"); @@ -237,7 +210,7 @@ private boolean isPruneAcknowledged() { return BooleanUtils.toBooleanDefaultIfNull(getPruneAcknowledged(), true); } - public void write(AdaptrisMessage msg) throws InterlokException { + void write(AdaptrisMessage msg) throws InterlokException { PreparedStatement ps = null; validateMessage(msg); Object[] params = new Object[] { msg.getUniqueId(), msg.getMetadataValue(Constants.ACKNOWLEDGE_ID_KEY), @@ -255,7 +228,7 @@ public void write(AdaptrisMessage msg) throws InterlokException { } } - public boolean delete(String msgId) throws InterlokException { + boolean delete(String msgId) throws InterlokException { PreparedStatement ps = null; Object[] params = new Object[] { msgId }; try { @@ -270,7 +243,7 @@ public boolean delete(String msgId) throws InterlokException { } } - public void acknowledge(String acknowledgeId) throws InterlokException { + void acknowledge(String acknowledgeId) throws InterlokException { PreparedStatement ps = null; Object[] params = { Constants.ACKNOWLEDGED, new Date(), acknowledgeId }; try { @@ -284,7 +257,7 @@ public void acknowledge(String acknowledgeId) throws InterlokException { } } - public void deleteAcknowledged() throws InterlokException { + void deleteAcknowledged() throws InterlokException { PreparedStatement ps = null; try { ps = prepareStatementWithoutParameters(sqlStatements.getProperty(DELETE_ACKNOWLEGED_SQL)); @@ -297,7 +270,7 @@ public void deleteAcknowledged() throws InterlokException { } } - public List obtainExpiredMessages() throws InterlokException { + List obtainExpiredMessages() throws InterlokException { List result = new ArrayList(); PreparedStatement ps = null; ResultSet rs = null; @@ -322,7 +295,7 @@ public List obtainExpiredMessages() throws InterlokException { return result; } - public List obtainMessagesToRetry() throws InterlokException { + List obtainMessagesToRetry() throws InterlokException { PreparedStatement ps = null; ResultSet rs = null; List result = new ArrayList(); @@ -352,7 +325,7 @@ public List obtainMessagesToRetry() throws InterlokException { return result; } - public void updateRetryCount(String messageId) throws InterlokException { + void updateRetryCount(String messageId) throws InterlokException { PreparedStatement ps = null; Object[] params = { new Date(), messageId }; try { @@ -425,13 +398,13 @@ private void createStoreTable() throws Exception { private PreparedStatement prepareStatementWithParameters(String query, Object[] parameters) throws SQLException { Connection conn; conn = makeConnection(); - PreparedStatement preparedStatement = conn.prepareStatement(query); + PreparedStatement ps = conn.prepareStatement(query); int count = 1; for (Object o : parameters) { - preparedStatement.setObject(count, o); + ps.setObject(count, o); count++; } - return preparedStatement; + return ps; } private PreparedStatement prepareStatementWithoutParameters(String query) throws SQLException { @@ -467,5 +440,4 @@ private Connection makeConnection() throws SQLException { return getConnection().retrieveConnection(DatabaseConnection.class).connect(); } - } diff --git a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryService.java b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryService.java index 7cffbed26..bf82c729f 100644 --- a/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryService.java +++ b/interlok-core/src/main/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryService.java @@ -15,12 +15,13 @@ import com.adaptris.core.Service; import com.adaptris.core.ServiceException; import com.adaptris.core.jdbc.retry.Constants; +import com.adaptris.core.util.LifecycleHelper; import com.adaptris.interlok.InterlokException; import com.thoughtworks.xstream.annotations.XStreamAlias; /** *

    - * Service which stores unacknowledged messages for future retry. + * Service which stores unacknowledged messages for future retry, with acknowledgment support. *

    *

    * This class supports both synchronous and asynchronous acknowledgement. This is controlled by the field @@ -80,28 +81,27 @@ public JdbcStoreMessageForRetryService() { setService(new NullService()); setAsyncAutoRetryOnFail(true); } - - /** @see com.adaptris.core.ServiceImp#start() */ + + /** @see com.adaptris.core.AdaptrisComponent#start() */ @Override - public void start() throws CoreException { - super.start(); - getService().start(); + protected void startService() throws CoreException { + LifecycleHelper.start(getService()); } - - /** @see com.adaptris.core.ServiceImp#stop() */ + + /** @see com.adaptris.core.AdaptrisComponent#stop() */ @Override - public void stop() { - getService().stop(); - super.stop(); + protected void stopService() { + LifecycleHelper.stop(getService()); } - /** - * - * @see JdbcRetryServiceImp#performService(com.adaptris.core.AdaptrisMessage) + * The main service method, which stores a message and details of the wrapped service in the retry database table. + * + * @see com.adaptris.core.Service#doService(com.adaptris.core.AdaptrisMessage) */ @Override - protected void performService(AdaptrisMessage msg) throws ServiceException { + public void doService(AdaptrisMessage msg) throws ServiceException { + pruneAcknowledged(); if (isAsynchronousAcknowledgment()) { handleAsynchronous(msg); } @@ -246,10 +246,4 @@ private boolean isAsyncAutoRetryOnFail() { return BooleanUtils.toBooleanDefaultIfNull(getAsyncAutoRetryOnFail(), true); } - @Override - protected void stopService() { - // TODO Auto-generated method stub - - } - } \ No newline at end of file From 292344e324c7b07e1e89f4af6fa69b3afa4bef96 Mon Sep 17 00:00:00 2001 From: James Wickham Date: Fri, 29 Sep 2023 18:04:16 +0100 Subject: [PATCH 10/10] writing unit tests --- .../retry/JdbcAcknowledgeServiceTest.java | 16 +- .../retry/JdbcRetryMessagesServiceTest.java | 16 +- .../jdbc/retry/JdbcRetryServiceCase.java | 517 ++++++++++++++++++ .../JdbcStoreMessageForRetryServiceTest.java | 14 +- 4 files changed, 535 insertions(+), 28 deletions(-) create mode 100644 interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceCase.java diff --git a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeServiceTest.java b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeServiceTest.java index b8c791987..ec5101aef 100644 --- a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeServiceTest.java +++ b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcAcknowledgeServiceTest.java @@ -1,14 +1,10 @@ package com.adaptris.core.services.jdbc.retry; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; - -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; - - -public class JdbcAcknowledgeServiceTest { +public class JdbcAcknowledgeServiceTest extends JdbcRetryServiceCase { + + @Override + protected JdbcAcknowledgeService createService() { + return new JdbcAcknowledgeService(); + } } \ No newline at end of file diff --git a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesServiceTest.java b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesServiceTest.java index d8bc5d15f..696736385 100644 --- a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesServiceTest.java +++ b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryMessagesServiceTest.java @@ -1,14 +1,10 @@ package com.adaptris.core.services.jdbc.retry; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; - -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; - - -public class JdbcRetryMessagesServiceTest { +public class JdbcRetryMessagesServiceTest extends JdbcRetryServiceCase { + + @Override + protected JdbcRetryMessagesService createService() { + return new JdbcRetryMessagesService(); + } } \ No newline at end of file diff --git a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceCase.java b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceCase.java new file mode 100644 index 000000000..81c01d384 --- /dev/null +++ b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcRetryServiceCase.java @@ -0,0 +1,517 @@ +package com.adaptris.core.services.jdbc.retry; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; + +import com.adaptris.core.AdaptrisConnection; +import com.adaptris.core.AdaptrisMessage; +import com.adaptris.core.AdaptrisMessageFactory; +import com.adaptris.core.CoreException; +import com.adaptris.core.MetadataElement; +import com.adaptris.core.NullConnection; +import com.adaptris.core.Service; +import com.adaptris.core.ServiceException; +import com.adaptris.core.jdbc.JdbcConnection; +import com.adaptris.core.jdbc.JdbcService; +import com.adaptris.core.services.exception.ConfiguredException; +import com.adaptris.core.services.exception.ThrowExceptionService; +import com.adaptris.core.services.metadata.AddMetadataService; +import com.adaptris.core.util.JdbcUtil; +import com.adaptris.interlok.junit.scaffolding.services.JdbcServiceCase; + +public abstract class JdbcRetryServiceCase extends JdbcServiceCase { + + protected static final String JDBC_RETRY_SERVICE_DRIVER = "jdbc.retryservice.driver"; + protected static final String JDBC_RETRY_SERVICE_URL = "jdbc.retryservice.url"; + protected static final String JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE = "jdbc.retryservice.sql.properties.file"; + + + @Test + public void testStoreMessageForRetrySyncNoFailure() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatPasses()); + execute(service, msg); + + + assertFalse(msg.headersContainsKey("retryAckId")); + assertFalse(msg.headersContainsKey("retryService")); + } + + @Test + public void testStoreMessageForRetrySyncWithFailure() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledge_id"); + + assertTrue(msg.headersContainsKey("retryAckId")); + assertTrue(msg.headersContainsKey("retryService")); + assertEquals(msg.getMetadataValue("retryAckId"), result); + } + + @Test + public void testStoreMessageForRetryAsyncAutoRetryOnFail() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(true); + service.setAsyncAutoRetryOnFail(true); + service.setService(createServiceForTestsThatPasses()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledge_id"); + + + assertTrue(msg.headersContainsKey("retryAckId")); + assertTrue(msg.headersContainsKey("retryService")); + assertEquals(msg.getMetadataValue("retryAckId"), result); + } + + @Test + public void testStoreMessageForRetryAsyncAutoRetryOnFailWithException() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(true); + service.setAsyncAutoRetryOnFail(true); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledge_id"); + + + assertTrue(msg.headersContainsKey("retryAckId")); + assertTrue(msg.headersContainsKey("retryService")); + assertEquals(msg.getMetadataValue("retryAckId"), result); + } + + @Test + public void testStoreMessageForRetryAsyncNoAutoRetryOnFail() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(true); + service.setAsyncAutoRetryOnFail(false); + service.setService(createServiceForTestsThatPasses()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledge_id"); + + assertTrue(msg.headersContainsKey("retryAckId")); + assertTrue(msg.headersContainsKey("retryService")); + assertEquals(msg.getMetadataValue("retryAckId"), result); + } + + @Test + public void testStoreMessageForRetryAsyncNoAutoRetryOnFailWithException() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(true); + service.setAsyncAutoRetryOnFail(false); + service.setService(createServiceForTestsThatFails()); + + + assertThrows(ServiceException.class, () -> { + execute(service, msg); + }, "Normal exception throwing."); + } + + @Test + public void testStoreMessageForRetryMissingRequiredMetadataException() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(true); + service.setAsyncAutoRetryOnFail(false); + service.setService(createServiceForTestsThatFails()); + + + assertThrows(ServiceException.class, () -> { + execute(service, msg); + }, "Required metadata missing"); + } + + @Test + public void testRetryMessageAsyncAutoRetryOnFail() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "40"); + msg.addMetadata("retryRetries", "3"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(true); + service.setAsyncAutoRetryOnFail(true); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "retries_to_date"); + + assertEquals("0", result.toString()); + + TimeUnit.MILLISECONDS.sleep(50); + + JdbcRetryMessagesService retryService = new JdbcRetryMessagesService(); + retryService.setConnection(createConnectionForService()); + retryService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + execute(retryService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "retries_to_date"); + + assertEquals("1", result.toString()); + } + + @Test + public void testRetryMessageAsyncNoAutoRetryOnFail() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "40"); + msg.addMetadata("retryRetries", "3"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(true); + service.setAsyncAutoRetryOnFail(false); + service.setService(createServiceForTestsThatPasses()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "retries_to_date"); + + assertEquals("0", result.toString()); + + TimeUnit.MILLISECONDS.sleep(50); + + JdbcRetryMessagesService retryService = new JdbcRetryMessagesService(); + retryService.setConnection(createConnectionForService()); + retryService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + execute(retryService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "retries_to_date"); + + assertEquals("1", result.toString()); + } + + @Test + public void testRetryMessageSyncWithException() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "40"); + msg.addMetadata("retryRetries", "3"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "retries_to_date"); + + assertEquals("0", result.toString()); + + TimeUnit.MILLISECONDS.sleep(50); + + JdbcRetryMessagesService retryService = new JdbcRetryMessagesService(); + retryService.setConnection(createConnectionForService()); + retryService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + execute(retryService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "retries_to_date"); + + assertEquals("1", result.toString()); + } + + + //TO REVIEW + @Test + public void testRetryMessageSync() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "40"); + msg.addMetadata("retryRetries", "-1"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "retries_to_date"); + + assertEquals("0", result.toString()); + + TimeUnit.MILLISECONDS.sleep(50); + + JdbcRetryMessagesService retryService = new JdbcRetryMessagesService(); + retryService.setConnection(createConnectionForService()); + retryService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + execute(retryService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "retries_to_date"); + + assertEquals("1", result.toString()); + } + + @Test + public void testRetryMessagePruneExpired() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "40"); + msg.addMetadata("retryRetries", "1"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledge_id"); + + TimeUnit.MILLISECONDS.sleep(50); + + JdbcRetryMessagesService retryService = new JdbcRetryMessagesService(); + retryService.setConnection(createConnectionForService()); + retryService.setPruneExpired(true); + retryService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + execute(retryService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledge_id"); + + assertEquals(msg.getMetadataValue("retryAckId"), result); + + TimeUnit.MILLISECONDS.sleep(50); + execute(retryService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledge_id"); + + assertEquals(null, result); + } + + @Test + public void testAcknowledgeWithoutPruning() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledged"); + + assertEquals("F", result); + + JdbcAcknowledgeService ackService = new JdbcAcknowledgeService(); + ackService.setConnection(createConnectionForService()); + ackService.setPruneAcknowledged(false); + ackService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + execute(ackService, msg); + execute(ackService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledged"); + + assertEquals("T", result); + } + + @Test + public void testAcknowledgeWithPruning() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledged"); + + assertEquals("F", result); + + JdbcAcknowledgeService ackService = new JdbcAcknowledgeService(); + ackService.setConnection(createConnectionForService()); + ackService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + execute(ackService, msg); + execute(ackService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledged"); + + assertEquals(null, result); + } + + public void testAcknowledgeWithNullAckId() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatFails()); + execute(service, msg); + Object[] params = new Object[] { msg.getUniqueId()}; + String result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledged"); + + assertEquals("F", result); + + JdbcAcknowledgeService ackService = new JdbcAcknowledgeService(); + ackService.setConnection(createConnectionForService()); + ackService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + msg.clearMetadata(); + msg.addMetadata("retryAckId", null); + execute(ackService, msg); + result = queryDatabaseForTests("SELECT * FROM retry_store WHERE message_id=?", params, "acknowledged"); + + assertEquals(null, result); + } + + @Test + public void testNullConnectionException() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + assertThrows(CoreException.class, () -> { + execute(service, msg); + }, "Null connection"); + } + + @Test + public void testNullExpiredMessageProducerException() throws Exception { + JdbcRetryMessagesService service = new JdbcRetryMessagesService(); + assertThrows(IllegalArgumentException.class, () -> { + service.setExpiredMessagesProducer(null); + }, "Null producer"); + } + + + @Test + public void testUnableToReadSqlPropertiesFile() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(createConnectionForService()); + service.setSqlPropertiesFile(PROPERTIES.getProperty("non existent")); + service.setAsynchronousAcknowledgment(true); + service.setAsyncAutoRetryOnFail(false); + service.setService(createServiceForTestsThatFails()); + + + assertThrows(CoreException.class, () -> { + execute(service, msg); + }, "Unable to read properties file"); + } + + @Test + public void testWrongConnectionType() throws Exception { + AdaptrisMessage msg = AdaptrisMessageFactory.getDefaultInstance().newMessage("test payload"); + msg.addMetadata("retryAckInterval", "2000"); + msg.addMetadata("retryRetries", "2"); + AdaptrisConnection nullConnection = new NullConnection(); + JdbcStoreMessageForRetryService service = new JdbcStoreMessageForRetryService(); + service.setConnection(nullConnection); + service.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + service.setAsynchronousAcknowledgment(false); + service.setService(createServiceForTestsThatFails()); + + JdbcRetryMessagesService retryService = new JdbcRetryMessagesService(); + retryService.setConnection(nullConnection); + retryService.setPruneExpired(true); + retryService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + + JdbcAcknowledgeService ackService = new JdbcAcknowledgeService(); + ackService.setConnection(nullConnection); + ackService.setSqlPropertiesFile(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_SQL_PROPERTIES_FILE)); + + assertThrows(ClassCastException.class, () -> { + execute(service, msg); + }, "Wrong connection type"); + assertThrows(ClassCastException.class, () -> { + execute(retryService, msg); + }, "Wrong connection type"); + + } + + + @Override + protected Object retrieveObjectForSampleConfig() { + return createService(); + } + + //Helper methods for the unit tests. + + protected abstract JdbcService createService(); + + private AdaptrisConnection createConnectionForService() { + return new JdbcConnection(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_URL), + PROPERTIES.getProperty(JDBC_RETRY_SERVICE_DRIVER)); + } + + private Service createServiceForTestsThatFails() { + return new ThrowExceptionService(new ConfiguredException("failed")); + } + + private Service createServiceForTestsThatPasses() { + return new AddMetadataService(new MetadataElement("key1", "val1")); + } + + + private Connection createConnection() throws Exception { + Class.forName(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_DRIVER)); + Connection conn = DriverManager.getConnection(PROPERTIES.getProperty(JDBC_RETRY_SERVICE_URL)); + conn.setAutoCommit(true); + return conn; + } + +//we query the same DB table outside of the service in order to check and compare entries were inserted, updated, deleted successfully +//and as expected. + private String queryDatabaseForTests(String query, Object[] parameters, String resultColumn) throws Exception { + Connection conn = createConnection(); + ResultSet rs = null; + PreparedStatement ps = conn.prepareStatement(query); + try { + int count = 1; + for (Object o : parameters) { + ps.setObject(count, o); + count++; + } + rs = ps.executeQuery(); + while (rs.next()) { + return rs.getObject(resultColumn).toString(); + } + } + finally { + JdbcUtil.closeQuietly(ps); + JdbcUtil.closeQuietly(conn); + } + return null; + } +} diff --git a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryServiceTest.java b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryServiceTest.java index b8ab3e174..c0a16bfdb 100644 --- a/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryServiceTest.java +++ b/interlok-core/src/test/java/com/adaptris/core/services/jdbc/retry/JdbcStoreMessageForRetryServiceTest.java @@ -1,14 +1,12 @@ package com.adaptris.core.services.jdbc.retry; -import static org.junit.jupiter.api.Assertions.assertFalse; -import static org.junit.jupiter.api.Assertions.assertNull; +import com.adaptris.core.jdbc.JdbcService; -import org.junit.jupiter.api.AfterAll; -import org.junit.jupiter.api.Assertions; -import org.junit.jupiter.api.BeforeAll; -import org.junit.jupiter.api.Test; +public class JdbcStoreMessageForRetryServiceTest extends JdbcRetryServiceCase { - -public class JdbcStoreMessageForRetryServiceTest { + @Override + protected JdbcService createService() { + return new JdbcStoreMessageForRetryService(); + } } \ No newline at end of file