Packages
riptide
0.2.74
0.5.2
0.5.1
0.5.0-beta9
0.5.0-beta8
0.5.0-beta7
0.5.0-beta6
0.5.0-beta5
0.5.0-beta4
0.5.0-beta3
0.5.0-beta2
0.5.0-beta11
0.5.0-beta10
0.5.0-beta
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.13
0.3.12
0.3.11
0.3.10
0.3.9
0.3.8
0.3.7
0.3.6
0.3.5
0.3.4
0.3.3
0.3.2
0.3.1
0.3.0
0.3.0-bd63a38
0.2.79
0.2.78
0.2.74
0.2.4
0.2.3
0.2.2
0.2.1
0.2.0
0.1.15
0.1.14
0.1.13
0.1.12
0.1.11
0.1.10
0.1.9
0.1.8
0.1.7
0.1.6
0.1.5
0.1.4
0.1.3
0.1.2
0.1.1
0.1.0
A data first framework for building realtime applications
Current section
Files
Jump to
Current section
Files
lib/riptide/store/store.ex
defmodule Riptide.Store do
@callback init(opts :: any()) :: :ok | {:error, atom()}
@callback mutation(merges :: any, deletes :: any(), opts :: any()) :: :ok | {:error, atom()}
@callback query(paths :: any, opts :: any()) :: any
def init() do
[
Riptide.Config.riptide_store_read(),
Riptide.Config.riptide_store_write()
]
|> Enum.uniq()
|> Enum.map(fn
{store, opts} -> :ok = store.init(opts)
_ -> :ok
end)
end
def mutation(mut) do
case Riptide.Config.riptide_store_write() do
{store, opts} ->
mutation(mut, store, opts)
_ ->
:ok
end
end
def mutation(mut, store, opts) do
merges = Dynamic.flatten(mut.merge)
deletes = Dynamic.flatten(mut.delete)
:ok = store.mutation(merges, deletes, opts)
end
def query(query) do
{store, opts} = Riptide.Config.riptide_store_read()
query(query, store, opts)
end
def query(query, store, store_opts) do
paths = query |> Riptide.Query.flatten() |> Enum.to_list()
paths
|> store.query(store_opts)
|> Stream.flat_map(fn {path, stream} ->
opts = Dynamic.get(query, path)
count = Enum.count(path)
stream
|> chunk(count, opts)
|> Stream.flat_map(fn values -> values end)
end)
|> inflate()
# paths
# |> Enum.reduce(%{}, fn {path, _}, collect ->
# value = Dynamic.get(result, path)
# Dynamic.put(collect, path, value)
# end)
end
def stream(path, opts \\ %{}) do
{store, store_opts} = Riptide.Config.riptide_store_read()
stream(path, opts, store, store_opts)
end
def stream(path, opts, store, store_opts) do
count = Enum.count(path)
[{path, opts}]
|> store.query(store_opts)
|> Stream.flat_map(fn {_path, stream} -> stream end)
|> chunk(count, opts)
|> Stream.map(fn values ->
values
|> Stream.map(fn {path, value} ->
{Enum.drop(path, count), value}
end)
end)
|> Stream.flat_map(&inflate/1)
end
# def chunk(stream, query) do
# chunked =
# Stream.chunk_while(stream, {nil, []}, fn {prefix, path, value}, {current, values} ->
# cond do
# current == nil -> {:cont, {prefix, [{path, value}]}}
# current == prefix -> {:cont, {prefix, [{path, value} | values]}}
# current !== prefix -> {:cont, {current, values}, {prefix, [{path, value}]}}
# end
# end)
# case opts[:limit] do
# nil -> chunked
# result -> Stream.take(chunked, result)
# end
# end
def chunk(stream, count, opts) do
chunked = Stream.chunk_by(stream, fn {path, _value} -> Enum.at(path, count) end)
case opts[:limit] do
nil -> chunked
result -> Stream.take(chunked, result)
end
end
def inflate(stream) do
stream
|> Enum.reduce(%{}, fn
{path, value}, collect when is_map(value) ->
Dynamic.combine(collect, Dynamic.put(%{}, path, value))
{path, value}, collect ->
Dynamic.put(collect, path, value)
end)
end
end