import java.sql.DriverManager
import java.util.concurrent.LinkedBlockingQueue
import java.util.zip.GZIPInputStream
import javax.xml.stream.XMLInputFactory
import javax.xml.stream.XMLStreamReader

class StaxCategory {
    static Object get(XMLStreamReader self, String key) {
        return self.getAttributeValue(null, key)
    }
    static String name(XMLStreamReader self) {
        return self.name.toString()
    }
    static String text(XMLStreamReader self) {
        return self.elementText
    }
}

url = "jdbc:hsqldb:hsql://localhost/xdb"
//url = "jdbc:hsqldb:file:event_db"
//url = "jdbc:hsqldb:mem:event_db"

Class.forName "org.hsqldb.jdbcDriver"

con = DriverManager.getConnection(url,"sa","")
con.createStatement().execute """
    CREATE CACHED TABLE log_event (
        id bigint primary key,
        attr_name varchar,
        attr_type varchar,
        attr_timestamp numeric,
        attr_id bigint
    );

    CREATE CACHED TABLE log_event_argument (
        id bigint primary key,
        attr_value varchar,
        attr_type varchar,
        attr_id bigint,
        log_event_id bigint
    );

    CREATE CACHED TABLE log_batch (
        id bigint primary key,
        attr_name varchar
    );

    CREATE CACHED TABLE log_transaction (
        id bigint primary key,
        attr_id bigint
    );

    CREATE CACHED TABLE log_batch_transaction (
        log_batch_id bigint,
        log_transaction_id bigint
    );

    CREATE CACHED TABLE log_event_batch (
        log_event_id bigint,
        log_batch_id bigint
    );
"""

logTransactionStmt = "INSERT INTO log_transaction (id,attr_id) VALUES (?,?)"
logBatchStmt = "INSERT INTO log_batch (id,attr_name) VALUES (?,?)"
logEventStmt = "INSERT INTO log_event (id,attr_name,attr_type,attr_timestamp,attr_id) VALUES (?,?,?,?,?)"
logEventArgumentStmt = "INSERT INTO log_event_argument (id,attr_value,attr_type,attr_id,log_event_id) VALUES (?,?,?,?,?)"
logEventBatchStmt = "INSERT INTO log_event_batch (log_event_id,log_batch_id) VALUES (?,?)"
logBatchTransactionStmt = "INSERT INTO log_batch_transaction (log_batch_id,log_transaction_id) VALUES (?,?)"

logTransactionId = 0
logBatchId = 0
logEventId = 0
logArgumentId = 0

inTransaction = false
inBatch = false

logTransactionQueue = new LinkedBlockingQueue()
logBatchQueue = new LinkedBlockingQueue()
logEventQueue = new LinkedBlockingQueue()
logEventArgumentQueue = new LinkedBlockingQueue()
logEventBatchQueue = new LinkedBlockingQueue()
logBatchTransactionQueue = new LinkedBlockingQueue()

batchsize = 10000

def consumer = { stmt, queue ->
    def con = DriverManager.getConnection(url,"sa","")
    con.autoCommit = false
    def ps = con.prepareStatement(stmt)
    def stmtCount = 0
    def argsList
    while (argsList = queue.take()) {
        def i = 1
        argsList.each {
            ps.setObject(i++, it == '' ? null : it)
        }
        ps.addBatch()
        stmtCount++
        if (stmtCount == batchsize) {
            ps.executeBatch()
            con.commit()
            ps.clearBatch()
            stmtCount = 0
            print " [Commit]"
        }
    }
    ps.executeBatch()
    con.commit()
}

Thread.start { consumer(logTransactionStmt, logTransactionQueue) }
Thread.start { consumer(logBatchStmt, logBatchQueue) }
Thread.start { consumer(logEventStmt, logEventQueue) }
Thread.start { consumer(logEventArgumentStmt, logEventArgumentQueue) }
Thread.start { consumer(logEventBatchStmt, logEventBatchQueue) }
Thread.start { consumer(logBatchTransactionStmt, logBatchTransactionQueue) }

Thread.start {
    while (true) {
        println """
=== Queue sizes ===
Transaction: ${logTransactionQueue.size()}
Batch:       ${logBatchQueue.size()}
Event:       ${logEventQueue.size()}
Argument:    ${logEventArgumentQueue.size()}

"""
        Thread.sleep 3000
    }

}

def processStartElement(element) {
    switch (element.name()) {

        case "transaction":
            inTransaction = true
            logTransactionId++
            logTransactionQueue.add ([logTransactionId, element.id])
            break

        case "batch":
            inBatch = true
            logBatchId++
            logBatchQueue.add ([logBatchId, element.name])
            if (inTransaction) {
                logBatchTransactionQueue.add ([logBatchId, logTransactionId])
            }
            break

        case "event":
            logEventId++
            logEventQueue.add ([logEventId,element.name,element.type,element.timestamp,element.id])
            if (inBatch) {
                logEventBatchQueue.add ([logEventId, logBatchId])
            }
            break

        case "argument":
            logArgumentId++
            logEventArgumentQueue.add ([logArgumentId,element.value,element.type,element.id,logEventId])
            break

    }
    //println "Start ${element.name()}"
}

def processEndElement(element) {
    switch (element.name()) {
        case "transaction":
            inTransaction = false
            break
        case "batch":
            inBatch = false
            break
    }
    //println "End ${element.name()}"
}

def insertEvents(source, desc) {
    println "Processing ${desc}..."

    reader = inputFactory.createXMLStreamReader (source)
    while (reader.hasNext()) {
        if (reader.startElement)
            processStartElement reader
        if (reader.endElement) {
            processEndElement reader
        }
        reader.next()
    }
    reader.close()
}

dir = "/Users/peter/jobb/assa_abloy/stoorn/"

inputFactory = XMLInputFactory.newInstance()

use (StaxCategory) {
    currentlog = new File(dir, "events\\currentlog.dat")
    insertEvents(currentlog.newInputStream(), "currentlog.dat")

    gzipFiles = new File(dir).listFiles({ file, name -> name.endsWith ".gz" } as FilenameFilter)
    gzipFiles.each {
        stream = new GZIPInputStream(it.newInputStream())
        if (it.size() < 100000 || true)
            insertEvents(stream, "${it.name} (${it.size()/1000} k)")
        else
            println "Skipping ${it.name} (too large)"
    }
}
queue.add(false)

println "DONE"