shackle/src/shackle.erl at 93e29acac58e45f2e49a6e64f4e753018f2e010b · lpgauth/shackle · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
-module(shackle).
-include("shackle_internal.hrl").

%% public
-export([
    call/2,
    call/3,
    cast/2,
    cast/3,
    cast/4,
    receive_response/1,
    receive_response/2
]).

%% types
-type cast() :: #cast {}.
-type cast_error() :: no_server | pool_not_started | shackle_not_started.
-type client() :: module().
-type external_request_id() :: term().
-type inet_address() :: inet:ip_address() | inet:hostname().
-type inet_port() :: inet:port_number().
-type protocol() :: shackle_socket | shackle_ssl | shackle_tcp | shackle_udp.
-type request_id() :: {shackle_server:name(), reference()}.
-type response() :: {external_request_id(), term()}.
-type socket() :: inet:socket() | socket:socket() | ssl:sslsocket().
-type socket_option() :: gen_tcp:connect_option() | gen_udp:option() | ssl:tls_client_option().
-type socket_options() :: [socket_option()].
-type table() :: atom().
-type time() :: pos_integer().

-export_type([
    cast/0,
    cast_error/0,
    client/0,
    external_request_id/0,
    inet_address/0,
    inet_port/0,
    protocol/0,
    request_id/0,
    response/0,
    socket/0,
    socket_options/0,
    table/0,
    time/0
]).

%% public
-spec call(shackle_pool:name(), term()) ->
    term() | {error, cast_error()}.

call(PoolName, Request) ->
    call(PoolName, Request, ?DEFAULT_TIMEOUT).

-spec call(atom(), term(), timeout()) ->
    term() | {error, cast_error()}.

call(PoolName, Request, Timeout) ->
    case cast(PoolName, Request, self(), Timeout) of
        {ok, RequestId} ->
            receive_response(RequestId);
        {error, Reason} ->
            {error, Reason}
    end.

-spec cast(shackle_pool:name(), term()) ->
    {ok, request_id()} | {error, cast_error()}.

cast(PoolName, Request) ->
    cast(PoolName, Request, self()).

-spec cast(shackle_pool:name(), term(), undefined | pid()) ->
    {ok, request_id()} | {error, cast_error()}.

cast(PoolName, Request, Pid) ->
    cast(PoolName, Request, Pid, ?DEFAULT_TIMEOUT).

-spec cast(shackle_pool:name(), term(), undefined | pid(), timeout()) ->
    {ok, request_id()} | {error, cast_error()}.

cast(PoolName, Request, Pid, Timeout) ->
    Timestamp = erlang:monotonic_time(microsecond),
    Ref = make_ref(),
    case shackle_pool:server(PoolName) of
        {ok, Client, Server} ->
            RequestId = {Server, Ref},
            Server ! {Request, #cast {
                client = Client,
                pid = Pid,
                request_id = RequestId,
                timeout = Timeout,
                timestamp = Timestamp
            }},
            {ok, RequestId};
        {error, Reason} ->
            {error, Reason}
    end.

-spec receive_response(request_id()) ->
    term() | {error, term()}.

receive_response(RequestId) ->
    receive
        {shackle_reply, RequestId, Reply} ->
            Reply
    end.

-spec receive_response(request_id(), timeout()) ->
    term() | {error, timeout | term()}.

receive_response(RequestId, Timeout) ->
    receive
        {shackle_reply, RequestId, Reply} ->
            Reply
    after Timeout ->
        {error, timeout}
    end.