Current section
Files
Jump to
Current section
Files
lib/marco_polo.ex
defmodule MarcoPolo do
@moduledoc """
Main API for interfacing with OrientDB.
This module provides functions to connect to a running OrientDB server and to
perform commands on it.
A connection to OrientDB can be established using the `start_link/1` function
and stoppped with `stop/1`.
## Connection type
OrientDB makes a distinction between *server operations* and *database
operations*. Server operations are operations that are performed at server
level: examples of these operations are checking if a database exists or
creating a new database. Database operations have to be performed inside a
database: examples of such operations are inserting a new record or getting
the number of records in the database.
Server and database operations can only be performed by the appropriate
connection: a connection to the server can perform **only** server operations,
while a connection to a database can perform **only** database operations. The
connection type is chosen when the connection is started via `start_link/1`.
## Examples
conn_type = {:db, "GratefulDeadConcerts", :document}
{:ok, conn} = MarcoPolo.start_link(user: "admin", password: "admin", connection: conn_type)
MarcoPolo.command(conn, "SELECT FROM OUser")
#=> {:ok, [...users...]}
"""
alias MarcoPolo.Connection, as: C
alias MarcoPolo.RID
alias MarcoPolo.Document
alias MarcoPolo.UndecodedDocument
alias MarcoPolo.BinaryRecord
alias MarcoPolo.Protocol
alias MarcoPolo.Protocol.RecordSerialization
@default_opts [
host: "localhost",
port: 2424,
]
@default_fetch_plan ""
@request_modes %{
sync: {:raw, <<0>>},
no_response: {:raw, <<2>>},
}
@tx_operation_types %{
update: 1,
delete: 2,
create: 3,
}
@type db_type :: :document | :graph
@type storage_type :: :plocal | :memory
@type rec :: Document.t | BinaryRecord.t
@type linked_records :: HashDict.t
@type tx_operation :: {:create | :update | :delete, rec}
@doc """
Starts the connection with an OrientDB server.
This function accepts the following options:
* `:user` - (string) the OrientDB user. This option is **required**.
* `:password` - (string) the OrientDB password. This option is **required**.
* `:connection` - specifies the connection type. This option is
**required**. To learn more about the connection type, refer to the docs
for the `MarcoPolo` module (there's a "Connection type" section). It can
be:
* `:server` - connects to the server to perform server operations
* `{:db, db_name, db_type}` - connects to a database to perform database
operations. `db_type` can be either `:document` or `:graph`.
* `:host` - (string or charlist) the host where the OrientDB server is
running. Defaults to `"localhost"`.
* `:port` - (integer) the port where the OrientDB server is running.
Defaults to `2424`.
It also accepts all options that `GenServer.start_link/3` accepts (e.g.,
`:name` for registering the new process or `:timeout` for providing a
connection timeout).
## Examples
Connecting to the server:
iex> {:ok, conn} = MarcoPolo.start_link user: "admin", password: "admin", connection: :server
iex> is_pid(conn)
true
Connecting to a database:
iex> connection = {:db, "MyDatabase", :document}
iex> {:ok, conn} = MarcoPolo.start_link user: "admin", password: "admin", connection: connection
iex> is_pid(conn)
true
"""
@spec start_link(Keyword.t) :: GenServer.on_start
def start_link(opts \\ []) do
C.start_link(Keyword.merge(@default_opts, opts))
end
@doc """
Closes the connection (asynchronously), doing the required cleanup work.
It always returns `:ok` as soon as it's called (regardless of the operation
being successful) since it is asynchronous.
## Examples
iex> MarcoPolo.stop(conn)
:ok
"""
@spec stop(pid) :: :ok
def stop(conn) do
C.stop(conn)
end
@doc """
Tells if the database called `name` with the given `type` exists.
This operation can only be performed on connections to the server. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Options
This function accepts the following options:
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
iex> MarcoPolo.db_exists?(conn, "GratefulDeadConcerts", :plocal)
{:ok, true}
"""
@spec db_exists?(pid, String.t, storage_type, Keyword.t) ::
{:ok, boolean} | {:error, term}
def db_exists?(conn, name, type, opts \\ []) when type in [:plocal, :memory] do
C.operation(conn, :db_exist, [name, Atom.to_string(type)], opts)
end
@doc """
Reloads the database to which `conn` is connected.
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Options
This function accepts the following options:
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
iex> MarcoPolo.db_reload(conn)
:ok
"""
@spec db_reload(pid) :: :ok | {:error, term}
def db_reload(conn, opts \\ []) do
case C.operation(conn, :db_reload, [], opts) do
{:ok, _} -> :ok
{:error, _} = error -> error
end
end
@doc """
Creates a database on the server.
`name` is used as the database name, `type` as the database type (`:document`
or `:graph`) and `storage` as the storage type (`:plocal` or `:memory`).
This operation can only be performed on connections to the server. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Options
This function accepts the following options:
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
iex> MarcoPolo.create_db(conn, "MyCoolDatabase", :document, :plocal)
:ok
"""
@spec create_db(pid, String.t, db_type, storage_type, Keyword.t) ::
:ok | {:error, term}
def create_db(conn, name, type, storage, opts \\ [])
when type in [:document, :graph] and storage in [:plocal, :memory] do
type = Atom.to_string(type)
storage = Atom.to_string(storage)
case C.operation(conn, :db_create, [name, type, storage], opts) do
{:ok, nil} -> :ok
o -> o
end
end
@doc """
Drop a database on the server.
This function drops the database identified by the name `name` and the storage
type `type` (either `:plocal` or `:memory`).
This operation can only be performed on connections to the server. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Options
This function accepts the following options:
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
iex> MarcoPolo.drop_db(conn, "UselessDatabase", :memory)
:ok
"""
@spec drop_db(pid, String.t, storage_type, Keyword.t) :: :ok | {:error, term}
def drop_db(conn, name, storage, opts \\ []) when storage in [:plocal, :memory] do
case C.operation(conn, :db_drop, [name, Atom.to_string(storage)], opts) do
{:ok, nil} -> :ok
o -> o
end
end
@doc """
Returns the size of the database to which `conn` is connected.
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Options
This function accepts the following options:
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
iex> MarcoPolo.db_size(conn)
{:ok, 1158891}
"""
@spec db_size(pid, Keyword.t) :: {:ok, non_neg_integer} | {:error, term}
def db_size(conn, opts \\ []) do
C.operation(conn, :db_size, [], opts)
end
@doc """
Returns the number of records in the database to which `conn` is connected.
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Options
This function accepts the following options:
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
iex> MarcoPolo.db_countrecords(conn)
{:ok, 7931}
"""
@spec db_countrecords(pid, Keyword.t) :: {:ok, non_neg_integer} | {:error, term}
def db_countrecords(conn, opts \\ []) do
C.operation(conn, :db_countrecords, [], opts)
end
@doc """
Creates a record in the database to which `conn` is connected.
`cluster_id` specifies the cluster to create the record in, while `record` is
the `MarcoPolo.Document` struct representing the record to create.
The return value in case of success is `{:ok, {rid, version}}` where `rid` is
the rid of the newly created record and `version` is the version of the newly
created record.
## Options
This function accepts the following options:
* `:no_response` - if `true`, send the request to the OrientDB server
without waiting for a response. This performs a *fire and forget*
operation, returning `:ok` every time.
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Examples
iex> record = %MarcoPolo.Document{class: "MyClass", fields: %{"foo" => "bar"}}
iex> MarcoPolo.create_record(conn, 15, record)
{:ok, {%MarcoPolo.RID{cluster_id: 15, position: 10}, 1}}
"""
@spec create_record(pid, non_neg_integer, rec, Keyword.t) ::
{:ok, {RID.t, non_neg_integer}} | {:error, term}
def create_record(conn, cluster_id, record, opts \\ []) do
args = [{:short, cluster_id}, record, record_type(record)]
if opts[:no_response] do
C.no_response_operation(conn, :record_create, args ++ [@request_modes.no_response])
else
refetching_schema conn, fn ->
C.operation(conn, :record_create, args ++ [@request_modes.sync], opts)
end
end
end
@doc """
Loads a record from the database to which `conn` is connected.
The record to load is identified by `rid`.
## Options
This function accepts the following options:
* `:fetch_plan` - the [fetching
strategy](http://orientdb.com/docs/last/Fetching-Strategies.html) used to
fetch the record from the database.
* `:ignore_cache` - if `true`, the cache is ignored, if `false` it's not.
Defaults to `true`.
* `:load_tombstones` - if `true`, information about deleted records is
loaded, if `false` it's not. Defaults to `false`.
* `:if_version_not_latest` - if `true`, only load the given record if the
version specified in the `:version` option is not the latest. If this
option is present, the `:version` option is required. This functionality
is supported in OrientDB >= 2.1.
* `:version` - see the `:if_version_not_latest` option.
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Return value
This function returns `{:ok, resp}` in case of a successful request or
`{:error, reason}` otherwise. In case of success, `resp` will be a two-element
tuple with the loaded record as the first elemen, and with a set of records
linked to it as the second element. This set of linked records can be
controlled via the `:fetch_plan` options. You're not supposed to manipulate
this value directly (so that the implementation can stay flexible); use the
functions in the `MarcoPolo.FetchPlan` module to work with linked records.
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Examples
iex> rid = %MarcoPolo.RID{cluster_id: 10, position: 184}
iex> {:ok, {record, linked_records}} = MarcoPolo.load_record(conn, rid)
iex> record.fields
%{"foo" => "bar"}
"""
@spec load_record(pid, RID.t, Keyword.t) :: {:ok, {rec, linked_records}} | {:error, term}
def load_record(conn, %RID{} = rid, opts \\ []) do
{op, args} =
if opts[:if_version_not_latest] do
args = [{:short, rid.cluster_id},
{:long, rid.position},
{:int, Keyword.fetch!(opts, :version)},
opts[:fetch_plan] || @default_fetch_plan,
opts[:ignore_cache] || true,
opts[:load_tombstones] || false]
{:record_load_if_version_not_latest, args}
else
args = [{:short, rid.cluster_id},
{:long, rid.position},
opts[:fetch_plan] || @default_fetch_plan,
opts[:ignore_cache] || true,
opts[:load_tombstones] || false]
{:record_load, args}
end
refetching_schema conn, fn ->
C.operation(conn, op, args, opts)
end
end
@doc """
Updates the given record in the databse to which `conn` is connected.
The record to update is identified by its `rid`; `version` is the version to
update. `new_record` is the updated record. `update_content?` can be:
* `true` - the content of the record has been changed and should be updated
in the storage.
* `false` - the record was modified but its own content has not changed:
related collections (e.g. RidBags) have to be updated, but the record
version and its contents should not be updated.
When the update is successful, `{:ok, new_version}` is returned; otherwise,
`{:error, reason}`.
## Options
This function accepts the following options:
* `:no_response` - if `true`, send the request to the OrientDB server
without waiting for a response. This performs a *fire and forget*
operation, returning `:ok` every time.
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Examples
iex> rid = %MarcoPolo.RID{cluster_id: 1, position: 10}
iex> new_record = %MarcoPolo.Document{class: "MyClass", fields: %{foo: "new value"}}
iex> MarcoPolo.update_record(conn, rid, 1, new_record, true)
{:ok, 2}
"""
@spec update_record(pid, RID.t, non_neg_integer, Document.t, boolean, Keyword.t) ::
{:ok, non_neg_integer} | {:error, term}
def update_record(conn, %RID{} = rid, version, new_record, update_content?, opts \\ []) do
args = [{:short, rid.cluster_id},
{:long, rid.position},
update_content?,
new_record,
version,
{:raw, "d"}]
if opts[:no_response] do
C.no_response_operation(conn, :record_update, args ++ [@request_modes.no_response])
else
C.operation(conn, :record_update, args ++ [@request_modes.sync], opts)
end
end
@doc """
Deletes a record from the database to which `conn` is connected.
The record to delete is identified by `rid`; version `version` is
deleted. Returns `{:ok, deleted?}` where `deleted?` is a boolean that tells if
the record has been deleted.
## Options
This function accepts the following options:
* `:no_response` - if `true`, send the request to the OrientDB server
without waiting for a response. This performs a *fire and forget*
operation, returning `:ok` every time.
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Examples
iex> rid = %MarcoPolo.RID{cluster_id: 76, position: 12}
iex> MarcoPolo.delete_record(conn, rid, 1)
{:ok, true}
"""
@spec delete_record(pid, RID.t, non_neg_integer, Keyword.t) ::
{:ok, boolean} | {:error, term}
def delete_record(conn, %RID{} = rid, version, opts \\ []) do
args = [{:short, rid.cluster_id},
{:long, rid.position},
{:int, version}]
if opts[:no_response] do
C.no_response_operation(conn, :record_delete, args ++ [@request_modes.no_response])
else
C.operation(conn, :record_delete, args ++ [@request_modes.sync], opts)
end
end
@doc """
Execute the given `query` in the database to which `conn` is connected.
OrientDB makes a distinction between idempotent queries and non-idempotent
queries (it calls the former *queries* and the latter *commands*). In order to
provide a clean interface for performing operations on the server, `MarcoPolo`
provides only a `command/3` function both for idempotent as well as
non-idempotent operations. Whether an operation is idempotent is inferred by
the text in `query`. As of now, `SELECT` and `TRAVERSE` operations are
idempotent while all other operations are non-idempotent.
## Options
The options that this function accepts depend in part on the type of the operation.
The options shared by both idempotent and non-idempotent operations are the following:
* `:params` - a map of params with atoms or strings as keys and any
encodable term as values. These parameters are used by OrientDB to build
prepared statements as you can see in the examples below. Defaults to `%{}`.
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
The additional options for idempotent (e.g., `SELECT`) queries are:
* `:fetch_plan`: a string specifying the fetch plan. Mandatory for `SELECT`
queries.
## Return value
If the query is successful then the return value is an `{:ok, values}`
tuple. `values` is a map with the following keys:
* `:response` - depends on the performed query. For example a `SELECT` query
will return a list of records, while a `CREATE CLUSTER` command will
return a cluster id.
* `:linked_records` - it's a set of additional records that have been
fetched by OrientDB. This can be controlled using a [fetch
plan](https://orientdb.com/docs/last/Fetching-Strategies.html) in the
query. You're not supposed to manipulate this value directly (so that the
implementation can stay flexible); use the functions in the
`MarcoPolo.FetchPlan` module to work with linked records.
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Examples
The following is an example of an idempotent command:
iex> opts = [params: %{name: "jennifer"}, fetch_plan: "*:-1"]
iex> query = "SELECT FROM User WHERE name = :name AND age > 18"
iex> {:ok, %MarcoPolo.Document{} = doc} = MarcoPolo.command(conn, query, opts)
iex> doc.fields["name"]
"jennifer"
iex> doc.fields["age"]
45
The following is an example of a non-idempotent command:
iex> query = "INSERT INTO User(name) VALUES ('meg', 'abed')"
iex> {:ok, [meg, abed]} = MarcoPolo.command(conn, query)
iex> meg.fields["name"]
"meg"
iex> abed.fields["name"]
"abed"
"""
@spec command(pid, String.t, Keyword.t)
:: {:ok, %{response: term, linked_records: linked_records}} | {:error, term}
def command(conn, query, opts \\ []) do
query_type = query_type(query)
command_class_name =
case query_type do
:sql_query -> "q"
:sql_command -> "c"
end
command_class_name = Protocol.Types.encode(command_class_name)
payload = encode_query_with_type(query_type, query, opts)
args = [{:raw, "s"}, # synchronous mode
IO.iodata_length([command_class_name, payload]),
{:raw, command_class_name},
{:raw, payload}]
refetching_schema conn, fn ->
C.operation(conn, :command, args, opts)
end
end
@doc """
Executes a script in the given `language` on the database `conn` is connected
to.
The text of the script is passed as `text`. `opts` is a list of options.
**Note**: for this to work, scripting must be enabled in the server
configuration. You can read more about scripting in the [OrientDB
docs](http://orientdb.com/docs/last/Javascript-Command.html#Enable_Server_side_scripting).
This operation can only be performed on connections to a database. To learn
more about the connection type, look at the "Connection type" section in the
docs for the `MarcoPolo` module.
## Options
This function accepts the following options:
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
iex> script = "for (i = 0; i < 3; i++) db.command('INSERT INTO Foo(idx) VALUES (' + i + ')');"
iex> {:ok, last_record} = MarcoPolo.script(conn, "Javascript", script)
iex> last_record.fields["idx"]
2
"""
@spec script(pid, String.t, String.t, Keyword.t) :: {:ok, term} | {:error, term}
def script(conn, language, text, opts \\ []) do
command_class_name = Protocol.Types.encode("s")
payload = [Protocol.Types.encode(language),
encode_query_with_type(:sql_command, text, opts)]
args = [{:raw, "s"}, # synchronous mode
IO.iodata_length([command_class_name, payload]),
{:raw, command_class_name},
{:raw, payload}]
refetching_schema conn, fn ->
C.operation(conn, :command, args, opts)
end
end
@doc """
Runs operations inside a transaction on the database to which `conn` is
connected.
This function will run a list of `operations` inside a server-side
transaction. Operations can be creations, updates and deletions of
records. Each operation has the form:
{op_type, record}
where `op_type` can be one of `:create`, `:update` or `:delete`.
## Options
This function accepts the following options:
* `:using_tx_log` - tells the server whether to use the transaction log to
recover the transaction or not. Defaults to `true`. *Note*: disabling the
log could speed up the execution of the transaction, but it makes
impossible to rollback the transaction in case of errors. This could lead
to inconsistencies in indexes as well, since in case of duplicated keys
the rollback is not called to restore the index status.
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
iex> ops = [{:create, %MarcoPolo.Document{class: "Foo", fields: %{"foo" => "bar"}}},
...> {:delete, %MarcoPolo.Document{rid: %MarcoPolo.RID{cluster_id: 1, position: 2}}]
iex> {:ok, %{created: [created], updated: []} = MarcoPolo.transaction(conn, ops)
iex> created
%MarcoPolo.Document{class: "Foo", fields: %{"foo" => "bar"}, rid: %MarcoPolo.RID{...}}
"""
@spec transaction(pid, [tx_operation], Keyword.t) ::
{:ok, %{created: [{RID.t, non_neg_integer}], updated: [{RID.t, non_neg_integer}]}} |
{:error, term}
def transaction(conn, operations, opts \\ []) when is_list(operations) do
{op_args, _index} = Enum.flat_map_reduce operations, -2, fn
{:create, _} = op, i ->
{args_from_tx_operation(op, i, opts), i - 1}
op, acc ->
{args_from_tx_operation(op, nil, opts), acc}
end
# The 0 at the end signals the end of the record list, while the empty
# binary is there because OrientDB :). It has to do with index changes, and
# it's not in the docs because OrientDB :).
args = [:transaction_id, opts[:using_tx_log] || true]
++ op_args
++ [{:raw, <<0>>}, <<>>]
C.operation(conn, :tx_commit, args, opts)
end
@doc """
Subscribes to a Live Query for the given `query`.
This function subscribes to a [Live
Query](https://orientdb.com/docs/last/Live-Query.html) for the given
`query`. Every time a change happens in the given query, a message will be
sent to `receiver`.
If the subscription is successful, this function returns `{:ok, token}` where
`token` is a unique identifier for the subscription to the give live
query. It's important to keep it around as it's needed to unsubscribe from the
live query (see `live_query_unsubscribe/2`).
The messages sent to `receiver` each time there's a change in the live query
has the following structure:
{:orientdb_live_query, token, {operation, record}}
where:
* `token` is the token mentioned above
* `operation` is one of `:create`, `:update`, or `:delete`, based on the
operation happened on the server
* `record` is the subject of the operation happened on the server
## Options
This function accepts the following options:
* `:timeout` - operation timeout in milliseconds. If this timeout expires,
an exit signal will be sent to the calling process.
## Examples
{:ok, token} = MarcoPolo.live_query(conn, "LIVE SELECT FROM Language", self())
Now, another client performs this query:
INSERT INTO Language(name, creators) VALUES ('Elixir', 'José Valim')
Back to the process that started the live query:
receive do
{:orientdb_live_query, ^token, {operation, record}} ->
operation #=> :create
record.fields["name"] #=> "Elixir"
end
"""
@spec live_query(pid, String.t, pid, Keyword.t) :: {:ok, integer} | {:error, term}
def live_query(conn, query, receiver, opts \\ []) do
command_class_name = Protocol.Types.encode("q") # always a query, never a command
payload = Protocol.Types.encode_list [
query,
-1,
opts[:fetch_plan] || @default_fetch_plan,
%Document{fields: %{"params" => %{}}},
]
args = [
{:raw, "l"}, # live query mode
IO.iodata_length([command_class_name, payload]),
{:raw, command_class_name},
{:raw, payload},
]
C.live_query(conn, args, receiver, opts)
end
@doc """
Unsubscribes from the live query identified by `token`.
This function unsubscribes from the live query identified by `token` (see
`live_query/4`). Once an unsubscription from a live query happens, no more
messages will be sent to the receiver specified when the live query had been
started.
This operation happens asynchronously, hence it always returns `:ok`.
## Examples
iex> MarcoPolo.live_query_unsubscribe(conn, token)
:ok
"""
@spec live_query_unsubscribe(pid, integer) :: :ok
def live_query_unsubscribe(conn, token) when is_integer(token) do
string_token = Integer.to_string(token)
case command(conn, "LIVE UNSUBSCRIBE #{string_token}") do
{:ok, %{response: %Document{fields: %{"unsubscribed" => ^string_token}}}} ->
C.live_query_unsubscribe(conn, token)
o ->
o
end
end
defp encode_query_with_type(:sql_query, query, opts) do
args = [query,
-1,
opts[:fetch_plan] || @default_fetch_plan,
%Document{class: nil, fields: %{"params" => to_params(opts[:params] || %{})}}]
Protocol.Types.encode_list(args)
end
defp encode_query_with_type(:sql_command, query, opts) do
args = [query]
if params = opts[:params] do
params = %Document{class: nil, fields: %{"parameters" => to_params(params)}}
# `true` means "use simple parameters".
args = args ++ [true, params]
else
args = args ++ [false]
end
args = args ++ [false]
Protocol.Types.encode_list(args)
end
defp refetching_schema(conn, fun) do
case fun.() do
{:ok, %{response: response, linked_records: linked} = r} ->
if unknown_property_ids?(response) or unknown_property_ids?(linked) do
schema = C.fetch_schema(conn)
response = redecode_with_new_schema(response, schema)
linked = redecode_with_new_schema(linked, schema)
{:ok, %{r | response: response, linked_records: linked}}
else
{:ok, r}
end
{:ok, {record, linked} = r} ->
if unknown_property_ids?(record) or unknown_property_ids?(linked) do
schema = C.fetch_schema(conn)
record = redecode_with_new_schema(record, schema)
linked = redecode_with_new_schema(linked, schema)
{:ok, {record, linked}}
else
{:ok, r}
end
o ->
o
end
end
defp unknown_property_ids?(%UndecodedDocument{}),
do: true
defp unknown_property_ids?(records) when is_list(records),
do: Enum.any?(records, &unknown_property_ids?/1)
defp unknown_property_ids?(%HashDict{} = records),
do: Enum.any?(records, &unknown_property_ids?/1)
defp unknown_property_ids?(_),
do: false
defp redecode_with_new_schema(records, schema) when is_list(records) do
Enum.map(records, &redecode_with_new_schema(&1, schema))
end
defp redecode_with_new_schema(%UndecodedDocument{rid: rid, version: vsn, content: content}, schema) do
doc = RecordSerialization.decode(content, schema)
%{doc | version: vsn, rid: rid}
end
defp redecode_with_new_schema(%HashDict{} = records, schema) do
for {rid, record} <- records, into: HashDict.new do
{rid, redecode_with_new_schema(record, schema)}
end
end
defp redecode_with_new_schema(record, _schema) do
record
end
defp to_params(params) when is_map(params) do
params
end
defp to_params(params) when is_list(params) do
params
|> Stream.with_index
|> Stream.map(fn({val, i}) -> {i, val} end)
|> Enum.into(%{})
end
defp query_type(query) do
case query_command(query) do
cmd when cmd in ["select", "traverse"] ->
:sql_query
_ ->
:sql_command
end
end
defp query_command(query) do
regex = ~r/^\s*(?<cmd>\w+)/
%{"cmd" => command} = Regex.named_captures(regex, query)
String.downcase(command)
end
defp args_from_tx_operation({:create, record}, incr_position, _opts) do
# -1 is the cluster id for the record that doesn't exist yet. OrientDB wants
# -it this way.
[
{:raw, <<1>>}, # continue to read records
{:raw, <<@tx_operation_types.create>>},
{:short, -1},
{:long, incr_position},
record_type(record),
record,
]
end
defp args_from_tx_operation({:update, %{__struct__: _, rid: %RID{} = rid} = record}, _incr_position, opts) do
unless record.version do
raise MarcoPolo.Error, "missing :version in the record #{inspect record}"
end
[
{:raw, <<1>>}, # continue to read records
{:raw, <<@tx_operation_types.update>>},
{:short, rid.cluster_id},
{:long, rid.position},
record_type(record),
{:int, record.version},
record,
opts[:update_content] || true,
]
end
defp args_from_tx_operation({:delete, %{__struct__: _, rid: %RID{} = rid} = record}, _incr_position, _opts) do
unless record.version do
raise MarcoPolo.Error, "missing :version in the record #{inspect record}"
end
[
{:raw, <<1>>}, # continue to read records
{:raw, <<@tx_operation_types.delete>>},
{:short, rid.cluster_id},
{:long, rid.position},
record_type(record),
{:int, record.version},
]
end
defp record_type(%Document{}), do: {:raw, "d"}
defp record_type(%BinaryRecord{}), do: {:raw, "b"}
end