<commit_before>/*
 * Copyright (C) FuseSource, Inc.
 * http://fusesource.com
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.fusesource.mq.leveldb

import org.apache.activemq.broker.BrokerService
import org.apache.activemq.broker.BrokerServiceAware
import org.apache.activemq.broker.ConnectionContext
import org.apache.activemq.command._
import org.apache.activemq.openwire.OpenWireFormat
import org.apache.activemq.usage.SystemUsage
import java.io.File
import java.io.IOException
import java.util.concurrent.ExecutionException
import java.util.concurrent.Future
import java.util.concurrent.atomic.AtomicLong
import util.Log
import reflect.BeanProperty
import org.apache.activemq.store._
import java.util._
import scala.collection.mutable.ListBuffer
import javax.management.ObjectName
import org.apache.activemq.broker.jmx.AnnotatedMBean
import org.apache.activemq.util._

object LevelDBStore extends Log {
  
  val DONE = new CountDownFuture();
  DONE.countDown
  
  def toIOException(e: Throwable): IOException = {
    if (e.isInstanceOf[ExecutionException]) {
      var cause: Throwable = (e.asInstanceOf[ExecutionException]).getCause
      if (cause.isInstanceOf[IOException]) {
        return cause.asInstanceOf[IOException]
      }
    }
    if (e.isInstanceOf[IOException]) {
      return e.asInstanceOf[IOException]
    }
    return IOExceptionSupport.create(e)
  }

  def waitOn(future: Future[AnyRef]): Unit = {
    try {
      future.get
    }
    catch {
      case e: Throwable => {
        throw toIOException(e)
      }
    }
  }
}

case class DurableSubscription(subKey:Long, topicKey:Long, info: SubscriptionInfo) {
  var lastAckPosition = 0L
  var cursorPosition = 0L
}

class LevelDBStoreView(val store:LevelDBStore) extends LevelDBStoreViewMBean {
  import store._

  def getAsyncBufferSize = asyncBufferSize
  def getIndexDirectory = directory.getCanonicalPath
  def getLogDirectory = Option(logDirectory).getOrElse(directory).getCanonicalPath
  def getIndexBlockRestartInterval = indexBlockRestartInterval
  def getIndexBlockSize = indexBlockSize
  def getIndexCacheSize = indexCacheSize
  def getIndexCompression = indexCompression
  def getIndexFactory = db.client.factory.getClass.getName
  def getIndexMaxOpenFiles = indexMaxOpenFiles
  def getIndexWriteBufferSize = indexWriteBufferSize
  def getLogSize = logSize
  def getParanoidChecks = paranoidChecks
  def getSync = sync
  def getVerifyChecksums = verifyChecksums

  def getUowClosedCounter = db.uowClosedCounter
  def getUowCanceledCounter = db.uowCanceledCounter
  def getUowStoringCounter = db.uowStoringCounter
  def getUowStoredCounter = db.uowStoredCounter

  def getUowMaxCompleteLatency = db.uow_complete_latency.get
  def getMaxIndexWriteLatency = db.client.max_index_write_latency.get
  def getMaxLogWriteLatency = db.client.log.max_log_write_latency.get
  def getMaxLogFlushLatency = db.client.log.max_log_flush_latency.get
  def getMaxLogRotateLatency = db.client.log.max_log_rotate_latency.get

  def resetUowMaxCompleteLatency = db.uow_complete_latency.reset
  def resetMaxIndexWriteLatency = db.client.max_index_write_latency.reset
  def resetMaxLogWriteLatency = db.client.log.max_log_write_latency.reset
  def resetMaxLogFlushLatency = db.client.log.max_log_flush_latency.reset
  def resetMaxLogRotateLatency = db.client.log.max_log_rotate_latency.reset

  def getIndexStats = db.client.index.getProperty("leveldb.stats")
}

class LevelDBStore extends ServiceSupport with BrokerServiceAware with PersistenceAdapter with TransactionStore {
  import LevelDBStore._

  final val wireFormat = new OpenWireFormat
  final val db = new DBManager(this)

  @BeanProperty
  var directory: File = null
  @BeanProperty
  var logDirectory: File = null
  
  @BeanProperty
  var logSize: Long = 1024 * 1024 * 100
  @BeanProperty
  var indexFactory: String = "org.fusesource.leveldbjni.JniDBFactory, org.iq80.leveldb.impl.Iq80DBFactory"
  @BeanProperty
  var sync: Boolean = true
  @BeanProperty
  var verifyChecksums: Boolean = false
  @BeanProperty
  var indexMaxOpenFiles: Int = 1000
  @BeanProperty
  var indexBlockRestartInterval: Int = 16
  @BeanProperty
  var paranoidChecks: Boolean = false
  @BeanProperty
  var indexWriteBufferSize: Int = 4 << 20
  @BeanProperty
  var indexBlockSize: Int = 4 * 1024
  @BeanProperty<commit_msg>Bump up the default index write buffer size<commit_after>/*
 * Copyright (C) FuseSource, Inc.
 * http://fusesource.com
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.fusesource.mq.leveldb

import org.apache.activemq.broker.BrokerService
import org.apache.activemq.broker.BrokerServiceAware
import org.apache.activemq.broker.ConnectionContext
import org.apache.activemq.command._
import org.apache.activemq.openwire.OpenWireFormat
import org.apache.activemq.usage.SystemUsage
import java.io.File
import java.io.IOException
import java.util.concurrent.ExecutionException
import java.util.concurrent.Future
import java.util.concurrent.atomic.AtomicLong
import util.Log
import reflect.BeanProperty
import org.apache.activemq.store._
import java.util._
import scala.collection.mutable.ListBuffer
import javax.management.ObjectName
import org.apache.activemq.broker.jmx.AnnotatedMBean
import org.apache.activemq.util._

object LevelDBStore extends Log {
  
  val DONE = new CountDownFuture();
  DONE.countDown
  
  def toIOException(e: Throwable): IOException = {
    if (e.isInstanceOf[ExecutionException]) {
      var cause: Throwable = (e.asInstanceOf[ExecutionException]).getCause
      if (cause.isInstanceOf[IOException]) {
        return cause.asInstanceOf[IOException]
      }
    }
    if (e.isInstanceOf[IOException]) {
      return e.asInstanceOf[IOException]
    }
    return IOExceptionSupport.create(e)
  }

  def waitOn(future: Future[AnyRef]): Unit = {
    try {
      future.get
    }
    catch {
      case e: Throwable => {
        throw toIOException(e)
      }
    }
  }
}

case class DurableSubscription(subKey:Long, topicKey:Long, info: SubscriptionInfo) {
  var lastAckPosition = 0L
  var cursorPosition = 0L
}

class LevelDBStoreView(val store:LevelDBStore) extends LevelDBStoreViewMBean {
  import store._

  def getAsyncBufferSize = asyncBufferSize
  def getIndexDirectory = directory.getCanonicalPath
  def getLogDirectory = Option(logDirectory).getOrElse(directory).getCanonicalPath
  def getIndexBlockRestartInterval = indexBlockRestartInterval
  def getIndexBlockSize = indexBlockSize
  def getIndexCacheSize = indexCacheSize
  def getIndexCompression = indexCompression
  def getIndexFactory = db.client.factory.getClass.getName
  def getIndexMaxOpenFiles = indexMaxOpenFiles
  def getIndexWriteBufferSize = indexWriteBufferSize
  def getLogSize = logSize
  def getParanoidChecks = paranoidChecks
  def getSync = sync
  def getVerifyChecksums = verifyChecksums

  def getUowClosedCounter = db.uowClosedCounter
  def getUowCanceledCounter = db.uowCanceledCounter
  def getUowStoringCounter = db.uowStoringCounter
  def getUowStoredCounter = db.uowStoredCounter

  def getUowMaxCompleteLatency = db.uow_complete_latency.get
  def getMaxIndexWriteLatency = db.client.max_index_write_latency.get
  def getMaxLogWriteLatency = db.client.log.max_log_write_latency.get
  def getMaxLogFlushLatency = db.client.log.max_log_flush_latency.get
  def getMaxLogRotateLatency = db.client.log.max_log_rotate_latency.get

  def resetUowMaxCompleteLatency = db.uow_complete_latency.reset
  def resetMaxIndexWriteLatency = db.client.max_index_write_latency.reset
  def resetMaxLogWriteLatency = db.client.log.max_log_write_latency.reset
  def resetMaxLogFlushLatency = db.client.log.max_log_flush_latency.reset
  def resetMaxLogRotateLatency = db.client.log.max_log_rotate_latency.reset

  def getIndexStats = db.client.index.getProperty("leveldb.stats")
}

class LevelDBStore extends ServiceSupport with BrokerServiceAware with PersistenceAdapter with TransactionStore {
  import LevelDBStore._

  final val wireFormat = new OpenWireFormat
  final val db = new DBManager(this)

  @BeanProperty
  var directory: File = null
  @BeanProperty
  var logDirectory: File = null
  
  @BeanProperty
  var logSize: Long = 1024 * 1024 * 100
  @BeanProperty
  var indexFactory: String = "org.fusesource.leveldbjni.JniDBFactory, org.iq80.leveldb.impl.Iq80DBFactory"
  @BeanProperty
  var sync: Boolean = true
  @BeanProperty
  var verifyChecksums: Boolean = false
  @BeanProperty
  var indexMaxOpenFiles: Int = 1000
  @BeanProperty
  var indexBlockRestartInterval: Int = 16
  @BeanProperty
  var paranoidChecks: Boolean = false
  @BeanProperty
  var indexWriteBufferSize: Int = 1024*1024*6
  @BeanProperty
  var indexBlockSize: Int = 4 * 1024
  @BeanProperty
{'fraction_non_alphanumeric': 0.07136980666392431, 'fraction_numerical': 0.00781571369806664, 'mean_word_length': 4.667249417249417, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5449518', 'n_tokens_mistral': 3009, 'n_tokens_neox': 2920, 'n_words': 713}