//设置package包名称以及导入依赖的类
package akka.stream.alpakka.ironmq

import java.util.UUID

import com.typesafe.config.{Config, ConfigFactory}
import org.scalatest.concurrent.ScalaFutures
import org.scalatest.{BeforeAndAfterEach, Notifying, Suite}

import scala.concurrent.ExecutionContext
import scala.util.hashing.MurmurHash3

trait IronMqFixture extends AkkaStreamFixture with BeforeAndAfterEach with ScalaFutures { _: Suite with Notifying =>

  private implicit val ec = ExecutionContext.global
  private var mutableIronMqClient = Option.empty[IronMqClient]

  def ironMqClient: IronMqClient =
    mutableIronMqClient.getOrElse(throw new IllegalStateException("The IronMqClient is not initialized"))

  override protected def initConfig(): Config =
    ConfigFactory.parseString(s"""akka.stream.alpakka.ironmq {
        |  credentials {
        |    project-id = "${MurmurHash3.stringHash(System.currentTimeMillis().toString)}"
        |  }
        |}
      """.stripMargin).withFallback(super.initConfig())

  override protected def beforeEach(): Unit = {
    super.beforeEach()
    mutableIronMqClient = Option(IronMqClient(IronMqSettings(config.getConfig("akka.stream.alpakka.ironmq"))))
  }

  override protected def afterEach(): Unit = {
    mutableIronMqClient = Option.empty
    super.afterEach()
  }

  def givenQueue(name: Queue.Name): Queue = {
    val created = ironMqClient.createQueue(name).futureValue
    note(s"Queue created: ${created.name.value}", Some(created))
    created
  }

  def givenQueue(): Queue =
    givenQueue(Queue.Name(s"test-${UUID.randomUUID()}"))
} 

//设置package包名称以及导入依赖的类
package org.cddb.store

import akka.actor.{Actor, ActorRef}
import akka.event.Logging

import scala.util.hashing.MurmurHash3

case class Range(start: Long, end: Long)

class Master(distributor: Distributor, hashFunc: (String => Long)) extends Actor {
  val log = Logging(context.system, this)

  override def receive = {
    case Init() =>
    case Insert(obj: StoreObject, caller: ActorRef) =>
      log.debug(s"Received Insert message: $obj")
      distributor.distribute(hashFunc(obj.key)) ! Insert(obj, caller)
    case Retrieve(key: String, caller: ActorRef) =>
      log.debug(s"Received Retrieve message: $key, caller: $caller")
      distributor.distribute(hashFunc(key)) ! Retrieve(key, caller)
    case Delete(key: String, caller: ActorRef) =>
      log.debug(s"Received Delete message: $key, caller: $caller")
      distributor.distribute(hashFunc(key)) ! Delete(key, caller)
    case _ => log.info("unknown message")
  }

}

class Distributor(f: (String => ActorRef)) {

  private val ranges = Array.fill[Range](1)(Range(0, Long.MaxValue))
  private val nodes = Array.fill[ActorRef](1)(f("test"))

  def distribute(hash: Long): ActorRef = nodes(0)

}

object Master {

  def hashFunc(str: String): Long = {
    val a = str.toCharArray.map(_.toByte)
    val hash1 = MurmurHash3.arrayHash(a)
    val hash2 = MurmurHash3.arrayHash(a.reverse)
    hash1 * hash2
  }

  def main(args: Array[String]): Unit = {
  }

} 

//设置package包名称以及导入依赖的类
package org.janzhou.hashing

import scala.util.hashing.MurmurHash3
import scala.collection.mutable.BitSet
import scala.reflect.ClassTag

class BloomFilter[T:ClassTag](
  val numBuckets: Int,
  val numHashFunctions: Int,
  val bits: BitSet
) {
  def this(numBuckets: Int, numHashFunctions: Int) = this(numBuckets, numHashFunctions, new BitSet(numBuckets))
  def this(numBuckets: Int) = this(numBuckets, 3)

  def hashValues(key: T) = {
    val hash1 = key.##
    val hash2 = math.abs(MurmurHash3.mixLast(0, hash1))

    for {
      i <- 0 to numHashFunctions
    } yield {
      val h = hash1 + i * hash2
      val nextHash = if (h < 0) ~h else h
      nextHash % numBuckets
    }
  }

  def containsHashes(hashes:Iterable[Int]):Boolean = {
    hashes.forall(i => bits.contains(i))
  }

  def contains(o: T): Boolean = {
    containsHashes(hashValues(o))
  }

  
  def optimallySized[T:ClassTag](expectedNumItems: Double, falsePositiveRate: Double): BloomFilter[T] = {
    val (buckets, funs) = optimalSize(expectedNumItems, falsePositiveRate)
    new BloomFilter[T](buckets, funs)
  }
} 

//设置package包名称以及导入依赖的类
package com.github.filosganga.ironmqrx

import akka.Done
import org.scalatest.concurrent.{Futures, ScalaFutures}
import org.scalatest.{Notifying, BeforeAndAfterAll, BeforeAndAfterEach, Suite}

import scala.concurrent.{ExecutionContext, Future}
import scala.util.hashing.MurmurHash3

trait IronMqFixture extends AkkaFixture with BeforeAndAfterEach with BeforeAndAfterAll with ScalaFutures {
  _: Suite with Notifying =>

  private implicit val ec = ExecutionContext.global
  private var queues: Set[Queue.Name] = Set.empty

  lazy val ironMqClient: IronMqClient = IronMqClient(actorSystem)

  override protected def afterEach(): Unit = {
    Future.sequence(queues.map(ironMqClient.deleteQueue)).futureValue
    queues = Set.empty
    super.afterEach()
  }

  override protected def afterAll(): Unit = {
    ironMqClient.close()
    super.afterAll()
  }

  def givenQueue(name: Queue.Name): Queue = {
    val created = ironMqClient.createQueue(name).futureValue
    note(s"Queue created: ${created.name.value}", Some(created))
    queues = queues + created.name
    created
  }

  def givenQueue(): Queue = {
    givenQueue(Queue.Name(s"test-${MurmurHash3.stringHash(System.currentTimeMillis().toString)}"))
  }
} 

//设置package包名称以及导入依赖的类
package com.avsystem.scex.util

import java.io.UnsupportedEncodingException
import java.nio.charset.StandardCharsets

import com.avsystem.commons.jiop.JavaInterop._
import com.avsystem.scex.presentation.annotation.Documentation
import org.apache.commons.codec.binary.{Base64, Hex}

import scala.collection.mutable
import scala.util.hashing.MurmurHash3

final class Bytes(val bytes: Array[Byte]) extends Comparable[Bytes] {
  override def hashCode(): Int = MurmurHash3.bytesHash(bytes)
  override def equals(other: Any): Boolean = other match {
    case b: Bytes => java.util.Arrays.equals(bytes, b.bytes)
    case _ => false
  }
  override def toString: String = s"Bytes($escaped)"
  override def compareTo(o: Bytes): Int = {
    def loop(i: Int): Int =
      if (i == bytes.length && i == o.bytes.length) 0
      else if (i == bytes.length) -1
      else if (i == o.bytes.length) 1
      else {
        val b1 = bytes(i)
        val b2 = o.bytes(i)
        if (b1 == b2) loop(i + 1) else b1 - b2
      }
    loop(0)
  }

  @Documentation("Encodes this sequence of bytes as string with non-ASCII bytes and backslash escaped, e.g. 'hsg\\x7c\\x0dfoo\\\\bar'")
  def escaped: String = EscapedBytes.render(bytes)

  @Documentation("Encodes this sequence of bytes as hexadecimal string")
  def hex: String = Hex.encodeHexString(bytes)

  @Documentation("Encodes this sequence of bytes as BASE64 string")
  def base64: String = Base64.encodeBase64String(bytes)

  @Documentation("Decodes this sequence of bytes as UTF-8 string")
  def decodeUTF8: String = new String(bytes, StandardCharsets.UTF_8)

  @Documentation("Decodes this sequence of bytes as string using given charset")
  def decode(charset: String): String =
    try new String(bytes, charset) catch {
      case e: UnsupportedEncodingException => throw new IllegalArgumentException(e)
    }

  def asList: JList[Byte] = new mutable.WrappedArray.ofByte(bytes).asJava
} 

//设置package包名称以及导入依赖的类
package teleporter.integration.utils

import scala.util.hashing.MurmurHash3


trait Hashing {
  val defaultHashCode: Int = -1
  val defaultHashByteArray: Array[Byte] = Bytes.toBytes(defaultHashCode)

  def hashValue(value: Any): Int = {
    val bytes = Option(value).map(Bytes.to).getOrElse(defaultHashByteArray)
    MurmurHash3.bytesHash(bytes)
  }

  def hash(seq: Traversable[Any]): Seq[Byte] = {
    seq.map(hashValue).flatMap(i ? Bytes.toBytes(i)).toSeq
  }

  def hash(m: Map[String, Any]): Seq[Byte] = {
    hash(m.values)
  }

  def hash(keys: Seq[String], m: Map[String, Any]): Seq[Byte] = {
    hash(keys.map(m(_)))
  }
}

object Hashing extends Hashing

trait ColumnFilter extends Hashing {
  def filterChanges(m: Map[String, Any], hashCodes: Array[Byte]): Map[String, Any] = {
    if (m.size * 4 != hashCodes.length) return m
    var i = -1
    m.filter { case (_, v) ? i += 1; hashValue(v) != Bytes.toInt(hashCodes.slice(i * 4, i * 4 + 4)) }
  }

  def filterChanges(keys: Seq[String], m: Map[String, Any], hashCodes: Array[Byte]): Map[String, Any] = {
    if (keys.size == hashCodes.length) return m
    var i = -1
    val filterKeys = keys.filter {
      key ? i += 1; m.getOrElse(key, defaultHashCode) != Bytes.toInt(hashCodes.slice(i * 4, i * 4 + 4))
    }.toSet
    m.filterKeys(filterKeys.contains)
  }
}

object ColumnFilter extends ColumnFilter