o Initial import.HEADmaster

2 files changed, 278 insertions, 0 deletions

diff --git a/src/test/java/io/trygvis/Main.java b/src/test/java/io/trygvis/Main.java
new file mode 100755
index 0000000..022d8cc
--- /dev/null
+++ b/src/test/java/io/trygvis/Main.java
@@ -0,0 +1,108 @@
+package io.trygvis;
+
+import io.trygvis.queue.*;
+import io.trygvis.queue.Queue;
+import org.slf4j.*;
+import org.slf4j.bridge.*;
+import org.springframework.beans.factory.annotation.*;
+import org.springframework.context.support.*;
+import org.springframework.stereotype.*;
+import org.springframework.transaction.*;
+import org.springframework.transaction.support.*;
+
+import java.util.*;
+
+import static java.lang.System.*;
+import static java.lang.Thread.*;
+
+@Component
+public class Main {
+/*
+    private static final Logger log = LoggerFactory.getLogger(Main.class);
+
+    public static void main(String[] args) throws Exception {
+        SLF4JBridgeHandler.install();
+
+        String username = getProperty("user.name");
+        setProperty("database.url", getProperty("jdbc.url", "jdbc:postgresql://localhost/" + username));
+        setProperty("database.username", username);
+        setProperty("database.password", username);
+//        setProperty("hibernate.showSql", "true");
+        setProperty("hibernate.hbm2ddl.auto", "create"); // create
+        setProperty("hibernate.dialect", PostgreSQL82Dialect.class.getName());
+
+        log.info("Starting context");
+        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("applicationContext.xml");
+        log.info("Started context");
+
+        try {
+            context.getBean(Main.class).run();
+            log.info("Sleeping");
+            sleep(1000 * 1000);
+        } catch (Exception e) {
+            e.printStackTrace(System.out);
+        }
+
+        log.info("Stopping context");
+        context.stop();
+        log.info("Stopped context");
+
+        exit(0);
+    }
+
+    @Autowired
+    private TransactionTemplate transactionTemplate;
+
+    @Autowired
+    private AsyncService asyncService;
+
+    @Autowired
+    @Qualifier("createArticle")
+    private AsyncService.AsyncCallable createArticleCallable;
+
+    @Autowired
+    @Qualifier("createArticle")
+    private AsyncService.AsyncCallable updateArticleCallable;
+
+    public void run() throws Exception {
+        log.info("Main.run");
+
+        final Queue q = asyncService.registerQueue("create-queue", 10, createArticleCallable);
+//        log.info("queue registered: ref = {}", q);
+//        asyncService.registerQueue("update-queue", 1, updateArticleCallable);
+
+//        q = asyncService.getQueue("create-queue");
+
+        final List<Task> tasks = new ArrayList<>();
+
+        transactionTemplate.execute(new TransactionCallbackWithoutResult() {
+            protected void doInTransactionWithoutResult(TransactionStatus status) {
+                for (int i = 0; i < 1; i++) {
+                    tasks.add(asyncService.schedule(q));
+                }
+            }
+        });
+
+        while (true) {
+            sleep(10000);
+
+            log.info("tasks.size = {}", tasks.size());
+            for (Iterator<Task> iterator = tasks.iterator(); iterator.hasNext(); ) {
+                Task task = iterator.next();
+
+                task = asyncService.update(task);
+
+                log.info("task = {}", task);
+
+                if (task.isDone()) {
+                    iterator.remove();
+                }
+            }
+
+            if (tasks.isEmpty()) {
+                break;
+            }
+        }
+    }
+*/
+}
diff --git a/src/test/java/io/trygvis/queue/QueueTest.java b/src/test/java/io/trygvis/queue/QueueTest.java
new file mode 100644
index 0000000..1aa77d7
--- /dev/null
+++ b/src/test/java/io/trygvis/queue/QueueTest.java
@@ -0,0 +1,170 @@
+package io.trygvis.queue;
+
+import com.jolbox.bonecp.*;
+import io.trygvis.queue.spring.*;
+import org.junit.*;
+import org.junit.runner.*;
+import org.slf4j.*;
+import org.slf4j.bridge.*;
+import org.springframework.beans.factory.annotation.*;
+import org.springframework.context.annotation.*;
+import org.springframework.context.support.*;
+import org.springframework.jdbc.core.*;
+import org.springframework.jdbc.datasource.*;
+import org.springframework.test.context.*;
+import org.springframework.test.context.junit4.*;
+import org.springframework.transaction.*;
+import org.springframework.transaction.support.*;
+
+import javax.sql.*;
+import java.util.*;
+
+import static java.lang.System.*;
+import static org.junit.Assert.*;
+import static org.springframework.transaction.TransactionDefinition.*;
+
+@SuppressWarnings("SpringJavaAutowiringInspection")
+@RunWith(SpringJUnit4ClassRunner.class)
+@ContextConfiguration(classes = TestConfig.class)
+public class QueueTest {
+
+    @Autowired
+    private TransactionTemplate transactionTemplate;
+
+    @Autowired
+    private JdbcTemplate jdbcTemplate;
+
+    @Autowired
+    private AsyncService asyncService;
+
+    private static Logger log;
+
+    @BeforeClass
+    public static void beforeClass() {
+        SLF4JBridgeHandler.install();
+
+        log = LoggerFactory.getLogger(QueueTest.class);
+
+        String username = getProperty("user.name");
+        setProperty("database.url", getProperty("jdbc.url", "jdbc:postgresql://localhost/" + username));
+        setProperty("database.username", username);
+        setProperty("database.password", username);
+    }
+
+    public int counter;
+    AsyncService.AsyncCallable callback = new AsyncService.AsyncCallable() {
+        public void run(List<String> args) throws Exception {
+            System.out.println("QueueTest.run");
+            counter++;
+        }
+    };
+
+    Queue queue;
+
+    @Before
+    public void before() {
+        queue = transactionTemplate.execute(new TransactionCallback<Queue>() {
+            public Queue doInTransaction(TransactionStatus status) {
+                return asyncService.registerQueue("wat", 10, callback);
+            }
+        });
+
+        int count = jdbcTemplate.update("DELETE FROM task");
+        log.info("Deleted {} tasks", count);
+    }
+
+    public void after() {
+        asyncService.stopQueue(queue);
+    }
+
+//    @Test
+//    public void testQueuePolling() {
+//        long start = currentTimeMillis();
+//        Task abc = transactionTemplate.execute(new TransactionCallback<Task>() {
+//            public Task doInTransaction(TransactionStatus status) {
+//                return asyncService.schedule(queue, "abc").getTask();
+//            }
+//        });
+//
+//        while(!abc.isDone()) {
+//            abc = asyncService.update(abc);
+//        }
+//
+//        long end = currentTimeMillis();
+//        log.info("Completed in {}ms", end - start);
+//
+//        assertEquals(1, counter);
+//    }
+
+    @Test
+    public void testQueueWaiting() throws Exception {
+        long start = currentTimeMillis();
+        TaskRef abc = transactionTemplate.execute(new TransactionCallback<TaskRef>() {
+            public TaskRef doInTransaction(TransactionStatus status) {
+                return asyncService.schedule(queue, "abc");
+            }
+        });
+
+        log.info("Waiting for signal");
+        abc.waitFor(1000);
+        long end = currentTimeMillis();
+        log.info("Completed in {}ms", end - start);
+
+        assertEquals(1, counter);
+    }
+}
+
+@Configuration
+@Import(value = QueueSpringConfig.class)
+class TestConfig {
+
+    @Bean
+    public static PropertySourcesPlaceholderConfigurer propertySourcesPlaceholderConfigurer() throws Exception {
+        return new PropertySourcesPlaceholderConfigurer() {{
+            setProperties(System.getProperties());
+            setLocalOverride(true);
+        }};
+    }
+
+    @Bean
+    public JdbcTemplate jdbcTemplate(DataSource dataSource) {
+        return new JdbcTemplate(dataSource);
+    }
+
+    @Bean
+    public DataSource dataSource(@Value("${database.url}") String jdbcUrl,
+                                 @Value("${database.username}") String username,
+                                 @Value("${database.password}") String password) {
+        BoneCPDataSource ds = new BoneCPDataSource();
+
+        ds.setLogStatementsEnabled(true);
+
+        ds.setJdbcUrl(jdbcUrl);
+        ds.setUsername(username);
+        ds.setPassword(password);
+
+        ds.setIdleConnectionTestPeriodInSeconds(60);
+        ds.setIdleMaxAgeInSeconds(240);
+        ds.setMaxConnectionsPerPartition(40);
+        ds.setMinConnectionsPerPartition(0);
+        ds.setPartitionCount(1);
+        ds.setAcquireIncrement(1);
+        ds.setStatementsCacheSize(1000);
+        ds.setReleaseHelperThreads(3);
+        ds.setStatisticsEnabled(true);
+        return new TransactionAwareDataSourceProxy(new LazyConnectionDataSourceProxy(ds));
+    }
+
+    @Bean
+    public PlatformTransactionManager transactionManager(DataSource dataSource) {
+        return new DataSourceTransactionManager(dataSource);
+    }
+
+    @Bean
+    public TransactionTemplate transactionTemplate(PlatformTransactionManager platformTransactionManager) {
+        DefaultTransactionDefinition td = new DefaultTransactionDefinition();
+        td.setPropagationBehavior(PROPAGATION_REQUIRED);
+        td.setIsolationLevel(ISOLATION_READ_COMMITTED);
+        return new TransactionTemplate(platformTransactionManager, td);
+    }
+}
\ No newline at end of file