package jnpf.message.mapper;
|
|
import jnpf.message.entity.MessageReceiveEntity;
|
import org.apache.ibatis.builder.xml.XMLMapperBuilder;
|
import org.apache.ibatis.datasource.unpooled.UnpooledDataSource;
|
import org.apache.ibatis.mapping.Environment;
|
import org.apache.ibatis.session.*;
|
import org.apache.ibatis.transaction.jdbc.JdbcTransactionFactory;
|
import org.junit.jupiter.api.Test;
|
import org.junit.jupiter.api.condition.EnabledIfEnvironmentVariable;
|
import java.io.StringReader;
|
import java.nio.charset.StandardCharsets;
|
import java.util.Objects;
|
import static org.junit.jupiter.api.Assertions.*;
|
|
@EnabledIfEnvironmentVariable(named="DMS_READ_TEST_URL", matches=".+")
|
class DmsUpgradeMessagePostgresTest {
|
@Test void receiptAndInboxRollbackTogetherAndDeduplicateAfterLostAcknowledgement() throws Exception {
|
var ds=new UnpooledDataSource("org.postgresql.Driver",System.getenv("DMS_READ_TEST_URL"),System.getenv("DMS_READ_TEST_USER"),System.getenv("DMS_READ_TEST_PASSWORD"));
|
var config=new Configuration(new Environment("message-test",new JdbcTransactionFactory(),ds));
|
String path="mapper/DmsUpgradeMessageMapper.xml";
|
try(var input=getClass().getResourceAsStream("/"+path)) {
|
String xml=new String(Objects.requireNonNull(input).readAllBytes(),StandardCharsets.UTF_8).replace("public.","pg_temp.");
|
new XMLMapperBuilder(new StringReader(xml),config,path,config.getSqlFragments()).parse();
|
}
|
try(var session=new SqlSessionFactoryBuilder().build(config).openSession(false)) {
|
try {
|
var c=session.getConnection();
|
try(var sql=c.createStatement()) {
|
sql.execute("CREATE TEMP TABLE base_msg_template (LIKE public.base_msg_template INCLUDING DEFAULTS) ON COMMIT DROP");
|
sql.execute("CREATE TEMP TABLE base_message (LIKE public.base_message INCLUDING DEFAULTS INCLUDING INDEXES) ON COMMIT DROP");
|
sql.execute("CREATE TEMP TABLE dms_template_upgrade_message_receipts (tenant_id text,task_id text,user_id text,created_at timestamp default CURRENT_TIMESTAMP,PRIMARY KEY(tenant_id,task_id,user_id)) ON COMMIT DROP");
|
sql.execute("INSERT INTO pg_temp.base_msg_template(f_id,f_tenant_id,f_en_code,f_message_type,f_message_source,f_enabled_mark,f_title,f_content) VALUES ('template','test','dmsTemplateUpgrade','1','3',1,'title','body')");
|
}
|
var mapper=session.getMapper(DmsUpgradeMessageMapper.class);
|
assertEquals("body",mapper.templates("test").getFirst().getContent()); assertTrue(mapper.templates("other").isEmpty());
|
var savepoint=c.setSavepoint();
|
assertEquals(1,mapper.claimDelivery("test","task","user"));
|
var message=new MessageReceiveEntity(); message.setId("m1");message.setUserId("user");message.setTitle("title");message.setBodyText("body");
|
assertEquals(1,mapper.insertMessage(message,"test"));
|
assertThrows(Exception.class,()->mapper.insertMessage(message,"test"));
|
c.rollback(savepoint);session.clearCache();
|
assertEquals(1,mapper.claimDelivery("test","task","user"));
|
assertEquals(1,mapper.insertMessage(message,"test"));
|
assertEquals(0,mapper.claimDelivery("test","task","user"));
|
assertEquals(1,mapper.claimDelivery("test","next-upgrade","user"));
|
assertEquals(1,mapper.claimDelivery("other","task","user"));
|
try(var sql=c.createStatement();var rows=sql.executeQuery("SELECT count(*) FROM pg_temp.base_message")) { assertTrue(rows.next());assertEquals(1,rows.getInt(1)); }
|
} finally { session.rollback(true); }
|
}
|
}
|
@Test void serviceTransactionRollsBackAllRecipientsAndAcknowledgesOnlyCommittedMessages() throws Exception {
|
try(var c=java.sql.DriverManager.getConnection(System.getenv("DMS_READ_TEST_URL"),System.getenv("DMS_READ_TEST_USER"),System.getenv("DMS_READ_TEST_PASSWORD"));
|
var pushes=org.mockito.Mockito.mockStatic(jnpf.message.util.PushMessageUtil.class)) {
|
var ds=new org.springframework.jdbc.datasource.SingleConnectionDataSource(c,true);
|
var config=new Configuration(new Environment("message-service-test",new org.mybatis.spring.transaction.SpringManagedTransactionFactory(),ds));
|
String path="mapper/DmsUpgradeMessageMapper.xml";
|
try(var input=getClass().getResourceAsStream("/"+path)) {
|
String xml=new String(Objects.requireNonNull(input).readAllBytes(),StandardCharsets.UTF_8).replace("public.","pg_temp.");
|
new XMLMapperBuilder(new StringReader(xml),config,path,config.getSqlFragments()).parse();
|
}
|
try(var sql=c.createStatement()) {
|
sql.execute("CREATE TEMP TABLE base_msg_template (LIKE public.base_msg_template INCLUDING DEFAULTS)");
|
sql.execute("CREATE TEMP TABLE base_message (LIKE public.base_message INCLUDING DEFAULTS INCLUDING INDEXES)");
|
sql.execute("ALTER TABLE pg_temp.base_message ADD CONSTRAINT test_failure CHECK(f_user_id != 'fail')");
|
sql.execute("CREATE TEMP TABLE dms_template_upgrade_message_receipts (tenant_id text,task_id text,user_id text,created_at timestamp default CURRENT_TIMESTAMP,PRIMARY KEY(tenant_id,task_id,user_id))");
|
sql.execute("INSERT INTO pg_temp.base_msg_template(f_id,f_tenant_id,f_en_code,f_message_type,f_message_source,f_enabled_mark,f_title,f_content) VALUES ('template','test','dmsTemplateUpgrade','1','3',1,'title','body')");
|
}
|
var session=new org.mybatis.spring.SqlSessionTemplate(new SqlSessionFactoryBuilder().build(config));
|
var target=new jnpf.message.service.DmsUpgradeMessageService(session.getMapper(DmsUpgradeMessageMapper.class));
|
var proxy=new org.springframework.aop.framework.ProxyFactory(target);
|
proxy.addAdvice(new org.springframework.transaction.interceptor.TransactionInterceptor(
|
new org.springframework.jdbc.datasource.DataSourceTransactionManager(ds),new org.springframework.transaction.annotation.AnnotationTransactionAttributeSource()));
|
var service=(jnpf.message.service.DmsUpgradeMessageService)proxy.getProxy();
|
var request=new jnpf.message.model.DmsUpgradeMessage();request.setTenantId("test");request.setTaskId("task");request.setActorId("actor");
|
request.setToUserIds(java.util.List.of("user","fail"));
|
assertThrows(Exception.class,()->service.send(request));
|
pushes.verifyNoInteractions();
|
try(var sql=c.createStatement();var rows=sql.executeQuery("SELECT (SELECT count(*) FROM pg_temp.base_message)+(SELECT count(*) FROM pg_temp.dms_template_upgrade_message_receipts)")) {
|
assertTrue(rows.next());assertEquals(0,rows.getInt(1));
|
}
|
request.setToUserIds(java.util.List.of("user")); assertTrue(service.send(request)); assertTrue(service.send(request));
|
pushes.verify(()->jnpf.message.util.PushMessageUtil.pushMessage(org.mockito.ArgumentMatchers.anyMap(),org.mockito.ArgumentMatchers.any(),org.mockito.ArgumentMatchers.eq(3)),org.mockito.Mockito.times(1));
|
try(var sql=c.createStatement();var rows=sql.executeQuery("SELECT count(*) FROM pg_temp.base_message")) { assertTrue(rows.next());assertEquals(1,rows.getInt(1)); }
|
}
|
}
|
|
}
|
