rjenkins

trait RiakStorageDriver[K, T] {
  def delete(t: T)
  def deleteByKey(key: K)
  def getByKey(key: K, converter: Converter[T]): Option[T]
  def persist(key: K, t: T, converter: Converter[T]): T
  def findFor2i(index: String, value: String, converter: Converter[T]): List[T]
  def findFor2i(index: String, value: Int, converter: Converter[T]): List[T]
  def deleteFor2i(index: String, value: String)
  def deleteFor2i(index: String, value: Int)
  def getBucket : String

rjenkins

class RiakDriver[T](bucket: String, riakClient: IRiakClient) 
  extends RiakStorageDriver[String, T] with Logging {

  def getByKey(key: String, converter: Converter[T]): Option[T] = {
    val data: IRiakObject = fetchBucket.fetch(key).execute()
    if (data != null)
      new Some(converter.toDomain(data))
    else
      None
  }

rjenkins

abstract class AbstractRiakEntityDAO[K, T](storageDriver: RiakStorageDriver[K,
  T]) extends GenericKeyValueDAO[K, T]
with Logging with Converter[T] {

  val stringIndexes = new ListBuffer[String]
  val integerIndexes = new ListBuffer[String]

  // Retrieve Entity by Key
  def getByKey(key: K): Option[T] = {
    storageDriver.getByKey(key, this)

rjenkins

  def findFor2i(index: String, value: String, converter: Converter[T]): List[T] = {
    get2iResults(getStringKeys(index, value), converter)
  }

  // Find all records of Type T for Matching 2i query
  private def getStringKeys(index: String, value: String): List[String] = {
    fetchBucket.fetchIndex(BinIndex.named(index)).withValue(value).execute().toList
  }

  private def get2iResults(keys: List[String], converter: Converter[T]): List[T] = {

rjenkins

case class Guitar(id: String, make: String, model: String, year: Int) {}

class GuitarDAO(storageDriver: RiakStorageDriver[String, Guitar])
  extends AbstractRiakEntityDAO[String, Guitar](storageDriver) with Converter[Guitar] {

  def fromDomain(guitar: Guitar, vClock: VClock): IRiakObject = {
    val dataAsString = generate(guitar)

    RiakObjectBuilder.newBuilder("guitars", guitar.id).withVClock(vClock)
      .withContentType(Constants.CTYPE_JSON)

rjenkins

trait GenericKeyValueDAO[K, T] {
  def persist(key: K, t: T): T
  def deleteByKey(key: K)
  def delete(t: T)
}

rjenkins

class RiakJSONEntityDAO[K, T <: PersistentEntity](storageDriver: RiakStorageDriver[K, T])(implicit mf: Manifest[T])
  extends AbstractRiakEntityDAO[K, T](storageDriver: RiakStorageDriver[K,
    T]) {


  def fromDomain(t: T, vClock: VClock): IRiakObject = {
    val dataAsString = generate(t)

    val iRiakObject = RiakObjectBuilder.newBuilder(storageDriver.getBucket, t.getKey)
      .withVClock(vClock)

rjenkins

case class Skateboard(id: String, brand: String, model: String, year: Int) 
extends PersistentEntity { def getKey=id }

class RiakJSONEntityDAOSpec extends Spec with Logging {

  val riakClient: IRiakClient = RiakFactory.pbcClient("localhost", 8087)
  val skateboardDao = new RiakJSONEntityDAO[String, Skateboard](new RiakDriver[Skateboard]("skateboards", riakClient))

  skateboardDao.addStringIndex("brand")

rjenkins

2012-08-03_18:06:36.69261 INFO  [2012-08-03 18:06:36,692] facade.PubSubService: Stream for 9JnrpKrMAxLfI11SypvQrx0cGEF-custom-query-8b49c9b1-78b8-4fea-b685-daf1d671146a-filter-d2870206-5958-40c7-a9f0-993b5cc808c3-aggregation-36fc43bf-66e5-4234-adc0-05b5d586b3b1-output is connected to ashbdrystrm05p.ood.ops, but should now consume (or re-subscribe) from ashbdrystrm07p.ood.ops. Switching...
2012-08-03_18:06:36.71975 ERROR [2012-08-03 18:06:36,715] facade.PubSubService: An error occurred in actor scalang.Process$$anon$7@5b3cb39a
2012-08-03_18:06:36.71978 ! scala.MatchError: (Reference('facade@ashbdrymess03p.ood.ops,Vector(6383, 1, 0),1),('state,9JnrpKrMAxLfI11SypvQrx0cGEF-discovery-output,java.nio.HeapByteBuffer[pos=0 lim=3064 cap=3064])) (of class scala.Tuple2)
2012-08-03_18:06:36.71979 !     at facade.PubSubService.handleInfo(PubSubService.scala:41)
2012-08-03_18:06:36.71980 !     at scalang.Service.onMessage(Service.scala:73)
2012-08-03_18:06:36.71980 !     at scalang.Process$$anon$7$$anonfun$onMessage$1.appl

rjenkins

# whatever flags you want to pass to the C compiler & linker
AM_CFLAGS = --pedantic -Wall -std=c99 -O2 
AM_LDFLAGS = -lm 

lib_LIBRARIES = libordacity.a
libordacity_a_SOURCES = Cluster.c queue_lock_mutex.c StringSet.c
 
include_HEADERS = ClusterConfig.h ClusterListener.h