Current section

7 Versions

Jump to

Compare versions

9 files changed
+209 additions
-176 deletions
  @@ -3,8 +3,9 @@
3 3 {<<"description">>,<<"An Elixir driver for Neo4J's bolt protocol.">>}.
4 4 {<<"elixir">>,<<"~> 1.3">>}.
5 5 {<<"files">>,
6 - [<<"mix.exs">>,<<"lib/boltex.ex">>,<<"lib/boltex/bolt.ex">>,
7 - <<"lib/boltex/error.ex">>,<<"lib/boltex/pack_stream.ex">>,
6 + [<<"mix.exs">>,<<"lib">>,<<"lib/boltex">>,<<"lib/boltex.ex">>,
7 + <<"lib/boltex/bolt.ex">>,<<"lib/boltex/error.ex">>,
8 + <<"lib/boltex/pack_stream">>,<<"lib/boltex/pack_stream.ex">>,
8 9 <<"lib/boltex/pack_stream/encode_error.ex">>,
9 10 <<"lib/boltex/pack_stream/encoder.ex">>,<<"lib/boltex/utils.ex">>,
10 11 <<"README.md">>,<<"LICENSE">>]}.
  @@ -15,4 +16,4 @@
15 16 {<<"maintainers">>,[<<"Michael Schaefermeyer">>]}.
16 17 {<<"name">>,<<"boltex">>}.
17 18 {<<"requirements">>,[]}.
18 - {<<"version">>,<<"0.4.0">>}.
19 + {<<"version">>,<<"0.4.1">>}.
  @@ -9,10 +9,10 @@ defmodule Boltex do
9 9 alias Boltex.Bolt
10 10
11 11 def test(host, port, query, params \\ %{}, auth \\ {}) do
12 - {:ok, p} = :gen_tcp.connect host, port, [active: false, mode: :binary, packet: :raw]
12 + {:ok, p} = :gen_tcp.connect(host, port, active: false, mode: :binary, packet: :raw)
13 13
14 - :ok = Bolt.handshake :gen_tcp, p
15 - {:ok, _info} = Bolt.init :gen_tcp, p, auth
14 + :ok = Bolt.handshake(:gen_tcp, p)
15 + {:ok, _info} = Bolt.init(:gen_tcp, p, auth)
16 16
17 17 Bolt.run_statement(:gen_tcp, p, query, params)
18 18 end
  @@ -2,27 +2,28 @@ defmodule Boltex.Bolt do
2 2 alias Boltex.{Utils, PackStream, Error}
3 3 require Logger
4 4
5 - @recv_timeout 10_000
6 - @max_chunk_size 65_535
5 + @recv_timeout 10_000
6 + @max_chunk_size 65_535
7 7
8 - @user_agent "Boltex/1.0"
9 - @hs_magic << 0x60, 0x60, 0xB0, 0x17 >>
10 - @hs_version << 1 :: 32, 0 :: 32, 0 :: 32, 0 :: 32 >>
8 + @user_agent "Boltex/0.4.1"
9 + @hs_magic <<0x60, 0x60, 0xB0, 0x17>>
10 + @hs_version <<1::32, 0::32, 0::32, 0::32>>
11 11
12 - @zero_chunk << 0, 0 >>
12 + def user_agent, do: @user_agent
13 13
14 - @sig_init 0x01
14 + @zero_chunk <<0, 0>>
15 +
16 + @sig_init 0x01
15 17 @sig_ack_failure 0x0E
16 - @sig_reset 0x0F
17 - @sig_run 0x10
18 - # @sig_discard_all 0x2F
19 - @sig_pull_all 0x3F
20 - @sig_success 0x70
21 - @sig_record 0x71
22 - @sig_ignored 0x7E
23 - @sig_failure 0x7F
18 + @sig_reset 0x0F
19 + @sig_run 0x10
20 + @sig_pull_all 0x3F
21 + @sig_success 0x70
22 + @sig_record 0x71
23 + @sig_ignored 0x7E
24 + @sig_failure 0x7F
24 25
25 - @summary ~w(success ignored failure)a
26 + @summary ~w(success ignored failure)a
26 27
27 28 @moduledoc """
28 29 The Boltex.Bolt module handles the Bolt protocol specific steps (i.e.
  @@ -48,11 +49,12 @@ defmodule Boltex.Bolt do
48 49 See "Shared options" in the documentation of this module.
49 50 """
50 51 def handshake(transport, port, options \\ []) do
51 - recv_timeout = get_recv_timeout options
52 + recv_timeout = get_recv_timeout(options)
53 +
54 + transport.send(port, @hs_magic <> @hs_version)
52 55
53 - transport.send port, @hs_magic <> @hs_version
54 56 case transport.recv(port, 4, recv_timeout) do
55 - {:ok, << 1 :: 32 >>} ->
57 + {:ok, <<1::32>>} ->
56 58 :ok
57 59
58 60 {:ok, other} ->
  @@ -82,8 +84,8 @@ defmodule Boltex.Bolt do
82 84 {:ok, info}
83 85 """
84 86 def init(transport, port, auth \\ {}, options \\ []) do
85 - params = auth_params auth
86 - send_messages transport, port, [{[@user_agent, params], @sig_init}]
87 + params = auth_params(auth)
88 + send_messages(transport, port, [{[@user_agent, params], @sig_init}])
87 89
88 90 case receive_data(transport, port, options) do
89 91 {:success, info} ->
  @@ -98,6 +100,7 @@ defmodule Boltex.Bolt do
98 100 end
99 101
100 102 defp auth_params({}), do: %{}
103 +
101 104 defp auth_params({username, password}) do
102 105 %{
103 106 scheme: "basic",
  @@ -114,7 +117,7 @@ defmodule Boltex.Bolt do
114 117 def send_messages(transport, port, messages) do
115 118 messages
116 119 |> encode_messages()
117 - |> Enum.each(&(transport.send(port, &1)))
120 + |> Enum.each(&transport.send(port, &1))
118 121 end
119 122
120 123 def encode_messages(messages) do
  @@ -124,32 +127,32 @@ defmodule Boltex.Bolt do
124 127 end
125 128
126 129 defp generate_binary_message({messages, signature}) do
127 - messages = List.wrap messages
128 - struct_size = length messages
130 + messages = List.wrap(messages)
131 + struct_size = length(messages)
129 132
130 - << 0xB :: 4, struct_size :: 4, signature >> <>
131 - Utils.reduce_to_binary(messages, &PackStream.encode/1)
133 + <<0xB::4, struct_size::4, signature>> <>
134 + Utils.reduce_to_binary(messages, &PackStream.encode/1)
132 135 end
133 136
134 137 defp generate_chunks(messages, chunks \\ [])
138 +
135 139 defp generate_chunks([], chunks) do
136 - Enum.reverse chunks
140 + Enum.reverse(chunks)
137 141 end
142 +
138 143 defp generate_chunks([message | messages], chunks)
139 - when byte_size(message) <= @max_chunk_size do
140 - message_size = byte_size message
141 - chunk =
142 - << message_size :: 16 >> <>
143 - message <>
144 - @zero_chunk
144 + when byte_size(message) <= @max_chunk_size do
145 + message_size = byte_size(message)
146 + chunk = <<message_size::16>> <> message <> @zero_chunk
145 147
146 - generate_chunks messages, [chunk | chunks]
148 + generate_chunks(messages, [chunk | chunks])
147 149 end
148 - defp generate_chunks([message | messages], chunks) do
149 - <<split :: binary-size(@max_chunk_size)>> <> rest = message
150 - chunk = << @max_chunk_size :: 16 >> <> split
151 150
152 - generate_chunks [rest | messages], [chunk | chunks]
151 + defp generate_chunks([message | messages], chunks) do
152 + <<split::binary-size(@max_chunk_size)>> <> rest = message
153 + chunk = <<@max_chunk_size::16>> <> split
154 +
155 + generate_chunks([rest | messages], [chunk | chunks])
153 156 end
154 157
155 158 @doc """
  @@ -175,23 +178,22 @@ defmodule Boltex.Bolt do
175 178 def run_statement(transport, port, statement, params \\ %{}, options \\ []) do
176 179 message = [statement, params]
177 180
178 - with :ok <- send_messages(transport, port, [{message, @sig_run}]),
181 + with :ok <- send_messages(transport, port, [{message, @sig_run}]),
179 182 {:success, _} = data <- receive_data(transport, port, options),
180 - :ok <- send_messages(transport, port, [{[], @sig_pull_all}]),
181 - more_data <- receive_data(transport, port, options),
182 - more_data = List.wrap(more_data),
183 - {:success, _} <- List.last(more_data)
184 - do
183 + :ok <- send_messages(transport, port, [{[], @sig_pull_all}]),
184 + more_data <- receive_data(transport, port, options),
185 + more_data = List.wrap(more_data),
186 + {:success, _} <- List.last(more_data) do
185 187 [data | more_data]
186 188 else
187 189 {:failure, map} ->
188 - Boltex.Error.exception map, port, :run_statement
190 + Boltex.Error.exception(map, port, :run_statement)
189 191
190 192 error = %Boltex.Error{} ->
191 193 error
192 194
193 195 error ->
194 - Boltex.Error.exception error, port, :run_statement
196 + Boltex.Error.exception(error, port, :run_statement)
195 197 end
196 198 end
197 199
  @@ -206,13 +208,13 @@ defmodule Boltex.Bolt do
206 208 See "Shared options" in the documentation of this module.
207 209 """
208 210 def ack_failure(transport, port, options \\ []) do
209 - send_messages transport, port, [
211 + send_messages(transport, port, [
210 212 {[], @sig_ack_failure}
211 - ]
213 + ])
212 214
213 215 case receive_data(transport, port, options) do
214 216 {:success, %{}} -> :ok
215 - error -> Boltex.Error.exception(error, port, :ack_failure)
217 + error -> Boltex.Error.exception(error, port, :ack_failure)
216 218 end
217 219 end
218 220
  @@ -227,13 +229,13 @@ defmodule Boltex.Bolt do
227 229 See "Shared options" in the documentation of this module.
228 230 """
229 231 def reset(transport, port, options \\ []) do
230 - send_messages transport, port, [
232 + send_messages(transport, port, [
231 233 {[], @sig_reset}
232 - ]
234 + ])
233 235
234 236 case receive_data(transport, port, options) do
235 237 {:success, %{}} -> :ok
236 - error -> Boltex.Error.exception(error, port, :reset)
238 + error -> Boltex.Error.exception(error, port, :reset)
237 239 end
238 240 end
239 241
  @@ -265,17 +267,16 @@ defmodule Boltex.Bolt do
265 267 See "Shared options" in the documentation of this module.
266 268 """
267 269 def receive_data(transport, port, options \\ [], previous \\ []) do
268 - with {:ok, data} <- do_receive_data(transport, port, options)
269 - do
270 + with {:ok, data} <- do_receive_data(transport, port, options) do
270 271 case unpack(data) do
271 272 {:record, _} = data ->
272 - receive_data transport, port, options, [data | previous]
273 + receive_data(transport, port, options, [data | previous])
273 274
274 275 {status, _} = data when status in @summary and previous == [] ->
275 276 data
276 277
277 278 {status, _} = data when status in @summary ->
278 - Enum.reverse [data | previous]
279 + Enum.reverse([data | previous])
279 280
280 281 other ->
281 282 {:error, Error.exception(other, port, :receive_data)}
  @@ -287,54 +288,55 @@ defmodule Boltex.Bolt do
287 288 end
288 289
289 290 defp do_receive_data(transport, port, options) do
290 - recv_timeout = get_recv_timeout options
291 + recv_timeout = get_recv_timeout(options)
291 292
292 293 case transport.recv(port, 2, recv_timeout) do
293 - {:ok, << chunk_size :: 16 >>} ->
294 - do_receive_data(transport, port, chunk_size, options, << >>)
294 + {:ok, <<chunk_size::16>>} ->
295 + do_receive_data(transport, port, chunk_size, options, <<>>)
295 296
296 297 other ->
297 298 other
298 299 end
299 300 end
300 - defp do_receive_data(transport, port, chunk_size, options, old_data) do
301 - recv_timeout = get_recv_timeout options
302 301
303 - with {:ok, data} <- transport.recv(port, chunk_size, recv_timeout),
304 - {:ok, marker} <- transport.recv(port, 2, recv_timeout)
305 - do
302 + defp do_receive_data(transport, port, chunk_size, options, old_data) do
303 + recv_timeout = get_recv_timeout(options)
304 +
305 + with {:ok, data} <- transport.recv(port, chunk_size, recv_timeout),
306 + {:ok, marker} <- transport.recv(port, 2, recv_timeout) do
306 307 case marker do
307 308 @zero_chunk ->
308 309 {:ok, old_data <> data}
309 310
310 - << chunk_size :: 16 >> ->
311 + <<chunk_size::16>> ->
311 312 data = old_data <> data
312 313 do_receive_data(transport, port, chunk_size, options, data)
313 314 end
314 315 else
315 316 other ->
316 - Error.exception other, port, :recv
317 + Error.exception(other, port, :recv)
317 318 end
318 319 end
319 320
320 321 @doc """
321 322 Unpacks (or in other words parses) a message.
322 323 """
323 - def unpack(<< 0x0B :: 4, packages :: 4, status, message :: binary >>) do
324 + def unpack(<<0x0B::4, packages::4, status, message::binary>>) do
324 325 response =
325 326 case PackStream.decode(message) do
326 327 response when packages == 1 ->
327 - List.first response
328 + List.first(response)
329 +
328 330 responses ->
329 331 responses
330 332 end
331 333
332 334 case status do
333 335 @sig_success -> {:success, response}
334 - @sig_record -> {:record, response}
336 + @sig_record -> {:record, response}
335 337 @sig_ignored -> {:ignored, response}
336 338 @sig_failure -> {:failure, response}
337 - other -> raise "Couldn't decode #{Utils.hex_encode << other >>}"
339 + other -> raise "Couldn't decode #{Utils.hex_encode(<<other>>)}"
338 340 end
339 341 end
  @@ -5,29 +5,29 @@ defmodule Boltex.Error do
5 5
6 6 def exception(%{"message" => message, "code" => code}, pid, function) do
7 7 %Boltex.Error{
8 - message: message,
9 - code: code,
8 + message: message,
9 + code: code,
10 10 connection_id: get_id(pid),
11 - function: function,
12 - type: :cypher_error,
11 + function: function,
12 + type: :cypher_error
13 13 }
14 14 end
15 15
16 16 def exception({:error, :closed}, pid, function) do
17 17 %Boltex.Error{
18 - message: "Port #{inspect pid} is closed",
18 + message: "Port #{inspect(pid)} is closed",
19 19 connection_id: get_id(pid),
20 - function: function,
21 - type: :connection_error,
20 + function: function,
21 + type: :connection_error
22 22 }
23 23 end
24 24
25 25 def exception(message, pid, function) do
26 26 %Boltex.Error{
27 - message: message_for(function, message),
27 + message: message_for(function, message),
28 28 connection_id: get_id(pid),
29 - function: function,
30 - type: :protocol_error,
29 + function: function,
30 + type: :protocol_error
31 31 }
32 32 end
33 33
  @@ -39,49 +39,57 @@ defmodule Boltex.Error do
39 39 instead of the Bolt Port (default: 7687).
40 40 """
41 41 end
42 +
42 43 defp message_for(:handshake, bin) when is_binary(bin) do
43 44 """
44 45 Handshake failed.
45 46 Expected 01:00:00:00 as a result, received: #{Utils.hex_encode(bin)}.
46 47 """
47 48 end
49 +
48 50 defp message_for(:hadshake, other) do
49 51 """
50 52 Handshake failed.
51 - Expected 01:00:00:00 as a result, received: #{inspect other}.
53 + Expected 01:00:00:00 as a result, received: #{inspect(other)}.
52 54 """
53 55 end
56 +
54 57 defp message_for(nil, message) do
55 58 """
56 - Unknown failure: #{inspect message}
59 + Unknown failure: #{inspect(message)}
57 60 """
58 61 end
62 +
59 63 defp message_for(_function, {:error, error}) do
60 - case error |> :inet.format_error |> to_string do
61 - "unknown POSIX error" -> to_string error
62 - other -> other
64 + case error |> :inet.format_error() |> to_string do
65 + "unknown POSIX error" -> to_string(error)
66 + other -> other
63 67 end
64 68 end
69 +
65 70 defp message_for(_function, {:ignored, []}) do
66 71 """
67 72 The session is in a failed state and ignores further messages. You need to
68 73 `ACK_FAILURE` or `RESET` in order to send new messages.
69 74 """
70 75 end
76 +
71 77 defp message_for(function, message) do
72 78 """
73 - #{function}: Unknown failure: #{inspect message}
79 + #{function}: Unknown failure: #{inspect(message)}
74 80 """
75 81 end
76 82
77 83 defp get_id({:sslsocket, {:gen_tcp, port, _tls, _unused_yet}, _pid}) do
78 84 get_id(port)
79 85 end
86 +
80 87 defp get_id(port) when is_port(port) do
81 88 case Port.info(port, :id) do
82 89 {:id, id} -> id
83 - nil -> nil
90 + nil -> nil
84 91 end
85 92 end
93 +
86 94 defp get_id(_), do: nil
87 95 end
  @@ -25,55 +25,60 @@ defmodule Boltex.PackStream do
25 25 # Decode
26 26 @doc "Decodes a binary stream recursively into Elixir data types"
27 27 # Null
28 - def decode(<< 0xC0, rest :: binary >>), do: [nil| decode(rest)]
28 + def decode(<<0xC0, rest::binary>>), do: [nil | decode(rest)]
29 29
30 30 # Boolean
31 - def decode(<< 0xC3, rest :: binary >>), do: [true | decode(rest)]
32 - def decode(<< 0xC2, rest :: binary >>), do: [false | decode(rest)]
31 + def decode(<<0xC3, rest::binary>>), do: [true | decode(rest)]
32 + def decode(<<0xC2, rest::binary>>), do: [false | decode(rest)]
33 33
34 34 # Float
35 - def decode(<< 0xC1, number :: float, rest :: binary >>) do
35 + def decode(<<0xC1, number::float, rest::binary>>) do
36 36 [number | decode(rest)]
37 37 end
38 38
39 39 # Strings
40 - def decode(<< 0x8 :: 4, str_length :: 4, rest :: bytes >>) do
41 - decode_text rest, str_length
40 + def decode(<<0x8::4, str_length::4, rest::bytes>>) do
41 + decode_text(rest, str_length)
42 42 end
43 - def decode(<< 0xD0, str_length, rest :: bytes >>) do
44 - decode_text rest, str_length
43 +
44 + def decode(<<0xD0, str_length, rest::bytes>>) do
45 + decode_text(rest, str_length)
45 46 end
46 - def decode(<< 0xD1, str_length :: 16, rest :: bytes >>) do
47 - decode_text rest, str_length
47 +
48 + def decode(<<0xD1, str_length::16, rest::bytes>>) do
49 + decode_text(rest, str_length)
48 50 end
49 - def decode(<< 0xD2, str_length :: 32, rest :: binary >>) do
50 - decode_text rest, str_length
51 +
52 + def decode(<<0xD2, str_length::32, rest::binary>>) do
53 + decode_text(rest, str_length)
51 54 end
52 55
53 56 # Lists
54 - def decode(<< 0x9 :: 4, list_size :: 4 >> <> bin), do: list(bin, list_size)
55 - def decode(<< 0xD4, list_size :: 8 >> <> bin), do: list(bin, list_size)
56 - def decode(<< 0xD5, list_size :: 16 >> <> bin), do: list(bin, list_size)
57 - def decode(<< 0xD6, list_size :: 32 >> <> bin), do: list(bin, list_size)
57 + def decode(<<0x9::4, list_size::4>> <> bin), do: list(bin, list_size)
58 + def decode(<<0xD4, list_size::8>> <> bin), do: list(bin, list_size)
59 + def decode(<<0xD5, list_size::16>> <> bin), do: list(bin, list_size)
60 + def decode(<<0xD6, list_size::32>> <> bin), do: list(bin, list_size)
58 61
59 62 # Maps
60 - def decode(<< 0xA :: 4, entries :: 4 >> <> bin), do: map(bin, entries)
61 - def decode(<< 0xD8, entries :: 8 >> <> bin), do: map(bin, entries)
62 - def decode(<< 0xD9, entries :: 16 >> <> bin), do: map(bin, entries)
63 - def decode(<< 0xDA, entries :: 32 >> <> bin), do: map(bin, entries)
63 + def decode(<<0xA::4, entries::4>> <> bin), do: map(bin, entries)
64 + def decode(<<0xD8, entries::8>> <> bin), do: map(bin, entries)
65 + def decode(<<0xD9, entries::16>> <> bin), do: map(bin, entries)
66 + def decode(<<0xDA, entries::32>> <> bin), do: map(bin, entries)
64 67
65 68 # Struct
66 - def decode(<< 0xB :: 4, struct_size :: 4, sig :: 8 >> <> struct) do
69 + def decode(<<0xB::4, struct_size::4, sig::8>> <> struct) do
67 70 {struct, rest} = struct |> decode |> Enum.split(struct_size)
68 71
69 72 [[sig: sig, fields: struct] | rest]
70 73 end
71 - def decode(<< 0xDC, struct_size :: 8, sig :: 8 >> <> struct) do
74 +
75 + def decode(<<0xDC, struct_size::8, sig::8>> <> struct) do
72 76 {struct, rest} = struct |> decode |> Enum.split(struct_size)
73 77
74 78 [[sig: sig, fields: struct] | rest]
75 79 end
76 - def decode(<< 0xDD, struct_size :: 16, sig :: 8 >> <> struct) do
80 +
81 + def decode(<<0xDD, struct_size::16, sig::8>> <> struct) do
77 82 {struct, rest} = struct |> decode |> Enum.split(struct_size)
78 83
79 84 [[sig: sig, fields: struct] | rest]
  @@ -83,14 +88,17 @@ defmodule Boltex.PackStream do
83 88 def decode(""), do: []
84 89
85 90 # Integers
86 - def decode(<< 0xC8, int, rest :: binary >>), do: [int | decode(rest)]
87 - def decode(<< 0xC9, int :: 16, rest :: binary >>), do: [int | decode(rest)]
88 - def decode(<< 0xCA, int :: 32, rest :: binary >>), do: [int | decode(rest)]
89 - def decode(<< 0xCB, int :: 64, rest :: binary >>), do: [int | decode(rest)]
90 - def decode(<< int, rest :: binary >>), do: [int | decode(rest)]
91 + def decode(<<0xC8, int::signed-integer, rest::binary>>), do: [int | decode(rest)]
92 + def decode(<<0xC9, int::signed-integer-16, rest::binary>>), do: [int | decode(rest)]
93 + def decode(<<0xCA, int::signed-integer-32, rest::binary>>), do: [int | decode(rest)]
94 + def decode(<<0xCB, int::signed-integer-64, rest::binary>>), do: [int | decode(rest)]
95 +
96 + def decode(<<int::signed-integer, rest::binary>>) do
97 + [int | decode(rest)]
98 + end
91 99
92 100 defp decode_text(bytes, str_length) do
93 - << string :: binary-size(str_length), rest :: binary >> = bytes
101 + <<string::binary-size(str_length), rest::binary>> = bytes
94 102
95 103 [string | decode(rest)]
96 104 end
  @@ -100,11 +108,12 @@ defmodule Boltex.PackStream do
100 108
101 109 [to_map(map) | rest]
102 110 end
111 +
103 112 defp to_map(map) do
104 113 map
105 114 |> Enum.chunk(2)
106 115 |> Enum.map(&List.to_tuple/1)
107 - |> Map.new
116 + |> Map.new()
108 117 end
109 118
110 119 defp list(list, list_size) do
Loading more files…