defmodule Kdb.Bucket do defstruct [:db, :name, :dbname, :handle, :t, :batch, :module, :ttl] @type t :: %__MODULE__{ db: reference(), name: atom(), dbname: atom(), handle: reference() | nil, t: :ets.tid() | nil, batch: reference() | nil, module: module(), ttl: boolean() } defmacro __using__(opts) do bucket = Keyword.get(opts, :name) || raise ArgumentError, "missing :name option" cache = Keyword.get(opts, :cache, Kdb.Cache) ttl = Keyword.get(opts, :ttl, true) decoder = Keyword.get(opts, :decoder, &Kdb.binary_to_term/1) encoder = Keyword.get(opts, :encoder, &Kdb.term_to_binary/1) quote bind_quoted: [ bucket: bucket, cache: cache, ttl: ttl, decoder: decoder, encoder: encoder ] do @bucket bucket @cache cache @ttl ttl @decoder decoder @encoder encoder @compile {:inline, put: 3, get: 2, has_key?: 2, delete: 2, incr: 3, batch: 2, decoder: 1, encoder: 1} def new(dbname, db, handle) do %Kdb.Bucket{ dbname: dbname, db: db, name: @bucket, handle: handle, t: new_table(), module: __MODULE__, ttl: @ttl } end def new_table do :ets.new(__MODULE__, [ :set, :public, read_concurrency: true, write_concurrency: true ]) end def name, do: @bucket def ttl, do: @ttl def decoder(x), do: @decoder.(x) def encoder(x), do: @encoder.(x) def batch(%Kdb.Bucket{dbname: dbname} = bucket, name) do batch = Kdb.batch(name) %{bucket | batch: batch} end def put(%Kdb.Bucket{batch: batch, handle: handle, t: t, ttl: ttl}, key, value) do :ets.insert(t, {key, value}) :rocksdb.batch_put(batch, handle, key, @encoder.(value)) ttl and @cache.put(@bucket, key) end defp put_in_memory(%Kdb.Bucket{t: t, ttl: ttl}, key, value) do :ets.insert(t, {key, value}) ttl and @cache.put(@bucket, key) end def get(bucket, key) do case :ets.lookup(bucket.t, key) do [{^key, value}] -> value [] -> get_from_disk(bucket, key) end end def has_key?(bucket, key) do case :ets.member(bucket.t, key) do true -> true false -> case get_from_disk(bucket, key) do nil -> false _value -> true end end end def get_from_disk(bucket, key) do case :rocksdb.get(bucket.db, bucket.handle, key, []) do {:ok, value} -> result = @decoder.(value) put_in_memory(bucket, key, result) result :not_found -> nil end end def incr(bucket, key, amount) when is_integer(amount) do result = get(bucket, key) || 0 result = :ets.update_counter(bucket.t, key, {2, amount}, {key, result}) :rocksdb.batch_put(bucket.batch, bucket.handle, key, @encoder.(result)) result end def delete(bucket, key) do :ets.delete(bucket.t, key) :rocksdb.batch_delete(bucket.batch, bucket.handle, key) @ttl and @cache.delete(@bucket, key) end end end defimpl Enumerable do def reduce(bucket, acc, fun) do # ← puedes parametrizar el nombre de tabla stream = Kdb.Stream.stream(bucket, decoder: &bucket.module.decoder/1) Enumerable.reduce(stream, acc, fun) end def count(bucket) do # No se puede contar sin recorrer todo {:error, bucket.module} end def member?(bucket, _element) do # No implementado de forma eficiente {:error, bucket.module} end def slice(bucket) do {:error, bucket.module} end end def fetch(bucket = %Kdb.Bucket{module: module}, key) do case :ets.lookup(bucket.t, key) do [{^key, value}] -> {:ok, value} [] -> case module.get_from_disk(bucket, key) do nil -> :error value -> {:ok, value} end end end def get_and_update(bucket = %Kdb.Bucket{module: module}, key, fun) do case module.get(bucket, key) do nil -> {:ok, nil, nil} value -> result = fun.(value) module.put(bucket, key, result) {:ok, value, result} end end defimpl Inspect do def inspect(bucket, _opts) do "#Kdb.Bucket" end end end defmodule Kdb.Stream do def stream(%Kdb.Bucket{db: db, handle: handle, module: module}, opts \\ []) do # <<>> or <<0>> or :last initial_seek = Keyword.get(opts, :seek, <<>>) # :next or :prev direction = Keyword.get(opts, :direction, :next) decoder_fun = Keyword.get(opts, :decoder, &module.decoder/1) Stream.resource( # Start: open iterator and seek fn -> {:ok, iter} = :rocksdb.iterator(db, handle, []) state = case :rocksdb.iterator_move(iter, initial_seek) do {:ok, key, value} -> {:ok, iter, key, value} _ -> {:done, iter} end state end, # Next: return {k, v} and move iterator fn {:done, iter} -> {:halt, iter} {:ok, iter, key, value} -> item = {key, decoder_fun.(value)} next = case :rocksdb.iterator_move(iter, direction) do {:ok, next_key, next_val} -> {:ok, iter, next_key, next_val} _ -> {:done, iter} end {[item], next} end, # After: close iterator fn iter -> :ok = :rocksdb.iterator_close(iter) end ) end end defmodule DefaultBucket do use Kdb.Bucket, name: :default, ttl: true end