defmodule Bolt.Sips.Client do @handshake_bytes_identifier <<0x60, 0x60, 0xB0, 0x17>> @noop_chunk <<0x00, 0x00>> alias Bolt.Sips.BoltProtocol.Versions alias Bolt.Sips.Utils.Converters alias Bolt.Sips.BoltProtocol.Message.{HelloMessage, InitMessage, LogonMessage} defstruct [:sock, :connection_id, :bolt_version] defmodule Config do @moduledoc false @default_timeout 5_000 defstruct [ :address, :port, :username, :password, :connect_timeout, :socket_options, :versions ] def new(opts) do address = Keyword.get(opts, :address, System.get_env("BOLT_HOST")) address = String.to_charlist(address || "localhost") versions = get_versions(opts) default_port = String.to_integer(System.get_env("BOLT_TCP_PORT") || "7687") port = Keyword.get(opts, :port, default_port) %__MODULE__{ address: address, port: port, username: Keyword.get(opts, :username, System.get_env("USER")) || raise(":username is missing"), password: Keyword.get(opts, :password), connect_timeout: Keyword.get(opts, :connect_timeout, @default_timeout), socket_options: Keyword.merge([mode: :binary, packet: :raw, active: false], opts[:socket_options] || []), versions: versions } end def get_versions(opts) do versions = case Keyword.get(opts, :versions) do nil -> case System.get_env("BOLT_VERSIONS") do nil -> Versions.latest_versions() env_versions -> env_versions |> String.split(",") |> Enum.map(&Converters.to_float/1) end ops_versions -> ops_versions end ((versions |> Enum.into([])) ++ [0, 0, 0]) |> Enum.take(4) |> Enum.sort(&>=/2) end end def connect(%Config{} = config) do with {:ok, client} <- do_connect(config) do handshake(client, config) end end def connect(opts) when is_list(opts) do connect(Config.new(opts)) end def do_connect(config) do %{ address: address, port: port, socket_options: socket_options, connect_timeout: connect_timeout } = config buffer? = Keyword.has_key?(socket_options, :buffer) client = %__MODULE__{connection_id: nil, sock: nil, bolt_version: nil} case :gen_tcp.connect(address, port, socket_options, connect_timeout) do {:ok, sock} when buffer? -> {:ok, %{client | sock: {:gen_tcp, sock}}} {:ok, sock} -> {:ok, [sndbuf: sndbuf, recbuf: recbuf, buffer: buffer]} = :inet.getopts(sock, [:sndbuf, :recbuf, :buffer]) buffer = buffer |> max(sndbuf) |> max(recbuf) :ok = :inet.setopts(sock, buffer: buffer) {:ok, %{client | sock: {:gen_tcp, sock}}} other -> other end end defp handshake(client, config) do case do_handshake(client, config) do {:ok, client} -> {:ok, client} {:error, reason} -> {:error, reason} end end defp do_handshake(client, config) do data = @handshake_bytes_identifier <> (config.versions |> Enum.sort(&>=/2) |> Enum.reduce(<<>>, fn version, acc -> acc <> Versions.to_bytes(version) end)) with :ok <- send_packet(client, data), encode_version <- recv_packets(client, config.connect_timeout), version <- decode_version(encode_version) do case version do 0.0 -> {:error, %Bolt.Sips.Error{code: :version_negotiation_error}} _ -> {:ok, %{client | bolt_version: version}} end else _ -> {:error, "Could not negotiate the version"} end end def message_hello(client, fields) do payload = HelloMessage.encode(client.bolt_version, fields) with :ok <- send_packet(client, payload) do recv_packets(client, &HelloMessage.decode/1, :infinity) end end def message_logon(client, fields) do payload = LogonMessage.encode(client.bolt_version, fields) with :ok <- send_packet(client, payload) do recv_packets(client, &LogonMessage.decode/1, :infinity) end end def message_init(client, fields) do payload = InitMessage.encode(client.bolt_version, fields) with :ok <- send_packet(client, payload) do recv_packets(client, &InitMessage.decode/1, :infinity) end end defp decode_version(<<0, 0, minor::unsigned-integer, major::unsigned-integer>>) when is_integer(major) and is_integer(minor) do Float.round(major + minor / 10.0, 1) end def send_packet(client, payload) do send_data(client, payload) end def send_data(%{sock: {sock_mod, sock}}, data) do sock_mod.send(sock, data) end def recv_packets(client, timeout) do case recv_data(client, timeout) do {:ok, response} -> response {:error, _} = error -> error end end def recv_packets(client, decoder, timeout) do recv_packets(client, decoder, timeout, <<>>) end defp decode_messages(response, chunks) do case response do @noop_chunk -> {:remaining_chunks, chunks} <<_::binary-size(byte_size(response)-2), 0,0>> -> <<_::16, message::binary>> = response {:complete_chunks, chunks <> message} << _::binary>> -> <<_::16, message::binary>> = response {:remaining_chunks, chunks <> message} end end defp recv_packets(client, decoder, timeout, chunks) do case recv_data(client, timeout) do {:ok, response} -> case decode_messages(response, chunks) do {:complete_chunks, binary_message} -> message = binary_message |> decoder.() case message do {:ok, decoded_message} -> {:ok, decoded_message} {:error, error_message} -> {:error, %Bolt.Sips.Error{code: error_message["code"], message: error_message["message"]}} end {:remaining_chunks, binary_message} -> recv_packets(client, decoder, timeout, binary_message) end {:error, _} = error -> error end end def recv_data(%{sock: {sock_mod, sock}}, timeout) do sock_mod.recv(sock, 0, timeout) end def disconnect(client) do {sock_mod, sock} = client.sock sock_mod.close(sock) :ok end def checkin(client) do {sock_mod, sock} = client.sock case sock_mod.setopts(sock, active: :once) do :ok -> :ok other -> other end end end