refactoring

Author: jlangch

doc/examples/scripts/ipc-wal-demo.venice

@@ -21,11 +21,13 @@
 
 ;;;; IPC Write-Ahead-Log demo
 
-(let [wal-dir (io/file (io/temp-dir "wal-"))]
+(let [srv-dir (io/file (io/temp-dir "srv-"))
+      wal-dir (io/file (io/temp-dir "wal-"))]
   (try
     ;; start client/server with Write-Ahead-Log and offer a few messages
     (println "Starting server/client ...")
     (try-with [server (ipc/server 33333
+                                  :server-log-dir srv-dir
                                   :write-ahead-log-dir wal-dir
                                   :write-ahead-log-compress true
                                   :write-ahead-log-compact true)
@@ -53,6 +55,7 @@
     ;; the new server will read the Write-Ahead-Logs and populate the queue :testq
     (println "Restarting server/client ...")
     (try-with [server (ipc/server 33333
+                                  :server-log-dir srv-dir
                                   :write-ahead-log-dir wal-dir
                                   :write-ahead-log-compress true
                                   :write-ahead-log-compact true)
@@ -79,6 +82,8 @@
 
     (sleep 100)
 
-    (finally (io/delete-file-tree wal-dir)))
+    (finally 
+       (io/delete-file-tree srv-dir)
+       (io/delete-file-tree wal-dir)))
 
   (println "Done."))

doc/examples/scripts/ipc-wal-load-demo.venice

@@ -21,14 +21,16 @@
 
 ;;;; IPC Write-Ahead-Log load demo
 
-(let [wal-dir       (io/file (io/temp-dir "wal-"))
+(let [srv-dir       (io/file (io/temp-dir "srv-"))
+      wal-dir       (io/file (io/temp-dir "wal-"))
       total-msgs     100_000
       half-msgs      (/ total-msgs 2)
       queue-capacity 1_000_000]    ;; capacity must be big enough
   (try
     ;; start client/server with Write-Ahead-Log and offer a few messages
     (println "Starting server/client ...")
     (try-with [server (ipc/server 33333
+                                  :server-log-dir srv-dir
                                   :write-ahead-log-dir wal-dir
                                   :write-ahead-log-compress true
                                   :write-ahead-log-compact true)
@@ -62,6 +64,7 @@
     ;; the new server will read the Write-Ahead-Logs and populate the queue :testq
     (println "Restarting server/client ...")
     (try-with [server (ipc/server 33333
+                                  :server-log-dir srv-dir
                                   :write-ahead-log-dir wal-dir
                                   :write-ahead-log-compress true
                                   :write-ahead-log-compact true)
@@ -84,6 +87,8 @@
 
     (sleep 100)
 
-    (finally (io/delete-file-tree wal-dir)))
+    (finally 
+      (io/delete-file-tree srv-dir)
+      (io/delete-file-tree wal-dir)))
 
   (println "Done."))

src/main/java/com/github/jlangch/venice/impl/functions/IPCFunctions.java

@@ -190,7 +190,7 @@ public VncVal apply(final VncList args) {
                 final File serverLogDir = serverLogDirVal == Nil
                                             ? null
                                             : IOFunctions.convertToFile(
-                                                walDirVal,
+                                                serverLogDirVal,
                                                 "Function 'ipc/server' arg ':server-log-dir' must be an `io/file`");
 
                 if (serverLogDir != null && !serverLogDir.isDirectory() && !serverLogDir.canWrite()) {

src/main/java/com/github/jlangch/venice/util/ipc/impl/wal/WalQueueManager.java

@@ -98,9 +98,12 @@ public Map<String, IpcQueue<Message>> preloadQueues()
 
         for(File logFile : listLogFiles()) {
             if (compactAtStart) {
+                logger.info(logFile, "WalQueueManager@preloadQueues: compacting WAL");
                 WriteAheadLog.compact(logFile, true, true);
             }
 
+            logger.info(logFile, "WalQueueManager@preloadQueues: create queue from WAL");
+
             final IpcQueue<Message> queue = DurableBoundedQueue.createFromWal(logFile, logger);
             queues.put(queue.name(), queue);
         };