(->request
{:as request-map,
:keys [url method query-string headers body response-consumer-factory request-options],
:or {method :get, headers default-headers}})
Bulk-chan takes a client, a partial request/option map, returns a
map of :input-ch
, :output-ch
and :flush-ch
.
:input-ch
is a channel that will accept bulk fragments to be
sent (either single or collection). It then will
wait (:flush-interval request-map) or (:flush-threshold request-map)
and then trigger an async request with the bulk payload accumulated.
Parallelism of the async requests is controllable via (:max-concurrent-requests request-map).
If the number of triggered requests exceeds the capacity of the job
buffer, puts! in :input-ch
will block (if done with async/put! you
can check the return value before overflowing the put! pending
queue).
Jobs results returned from the processing are a pair of job and responses map, or exception. The :output-ch will allow you to inspect [job responses] the server returned and handle potential errors/failures accordingly (retrying etc).
If you close! the :input-ch
it will close the underlying resources
and exit cleanly (consuming all jobs that remain in queues).
By default requests are run against _bulk, but the option map is passed as is to request-chan, you can overwrite options here and provide your own url, headers and so on.
:flush-ch
is provided to afford an immediate flush of the contents
of the job buffer by putting anything onto this channel.
source
(chunks->body chunks)
Utility function to create _bulk/_msearch bodies. It takes a
sequence of clj fragments and returns a newline delimited string of
JSON fragments
source
(client)
(client options)
Returns a client instance to be used to perform requests.
Options:
-
:hosts
: Collection of URIs of nodes - Defaults to ["http://localhost:9200"] -
:default-headers
: Sets the default request headers, which will be sent along with each request. Request-time headers will always overwrite any default headers. -
:failure-listener
: Sets the RestClient.FailureListener to be notified for each request failure -
:request
: request scoped config - extendable via theqbits.client-options/set-request-option!
multimethod):authentication?
:circular-redirect-allowed?
:connect-timeout
:connection-request-timeout
:content-compression?
:expect-continue?
:local-address
:max-redirects
:proxy
:redirects?
:relative-redirects-allowed?
:socket-timeout
:target-preferred-auth-schemes
-
:http-client
: http-client scoped config - extendable via theqbits.client-options/set-http-client-option!
multimethod):max-conn-per-route
:max-conn-total
:proxy
:ssl-context
:ssl-noop-hostname-verifier?
:user-agent
:auth-caching?
:cookie-management?
:basic-auth
(map of:user
:password
)
If you need extra/custom building you can hook into the builder by
extending the multimethod
qbits.spandex.client-options/set-option!
source
(close! this)
(decode-exception x)
Controls how to translate a client exception
source
(default-exception-handler ex)
Exception handler that will try to decode Exceptions via
ExceptionDecoder/decode-exception. If after decoding we still have a
throwable it will rethrow, otherwise it will pass on the value
returned. You can then extend ExceptionDecoder/decode-exception to
do whatever you'd like.
source
(encode-body x)
(node-selector f)
Creates a NodeSelector instance that will call f
as select():
see: https://github.com/elastic/elasticsearch/blob/master/client/rest/src/main/java/org/elasticsearch/client/NodeSelector.java#L29
source
(raw body)
Marks body as raw, which allows to skip JSON encoding
source
(request
client
{:keys [method url headers query-string body keywordize? response-consumer-factory exception-handler],
:or
{method :get,
keywordize? true,
exception-handler default-exception-handler,
response-consumer-factory HttpAsyncResponseConsumerFactory/DEFAULT},
:as request-params})
(request-async
client
{:keys [method url headers query-string body success error keywordize? response-consumer-factory exception-handler],
:or
{method :get,
keywordize? true,
exception-handler decode-exception,
response-consumer-factory HttpAsyncResponseConsumerFactory/DEFAULT},
:as request-params})
Similar to qbits.spandex/request
but returns immediately and works
asynchronously and triggers option :success
once a results is
received, or :error
if it was a failure
source
(request-chan client {:as options, :keys [ch]})
Similar to qbits.spandex/request
but runs asynchronously and
returns a core.async/promise-chan
that will have the result (or
error) delivered upon reception
source
(response-ex->ex-info re)
Utility function to transform an ResponseException into an ex-info
source
(response-ex->response re)
Return the response-map wrapped in a ResponseException
source
(scroll-chan client {:as request-map, :keys [ttl output-ch], :or {ttl "1m"}})
Returns a core async channel. Takes the same args as
qbits.spandex/request
. Perform async scrolling requests for a
query, request will be done as the user takes from the
channel. Every take! will request/return a page from the
scroll. You can specify scroll :ttl in the request map otherwise
it'll default to 1m. The chan will be closed once scroll is
complete. If you must stop scrolling before that, you must
async/close! manually, this will release all used resources. You can
also supply a :output-ch key to the request map, a core.async/chan that
will receive the results. This allow you to have custom buffers, or
have multiple scroll-chan calls feed the same channel instance
source
(set-sniff-on-failure! sniffer)
Register a SniffOnFailureListener that allows to perform sniffing
on failure.
source
(sniffer client)
(sniffer
client
{:as options,
:keys [scheme timeout],
:or {scheme :http, timeout ElasticsearchNodesSniffer/DEFAULT_SNIFF_REQUEST_TIMEOUT}})
Takes a Client instance (and possible sniffer options) and returns a sniffer instance that will initially be bound to passed client. Options:
-
:sniff-interval
: Sets the interval between consecutive ordinary sniff executions in milliseconds. Will be honoured when sniffOnFailure is disabled or when there are no failures between consecutive sniff executions. -
:sniff-after-failure-delay
: Sets the delay of a sniff execution scheduled after a failure (in milliseconds)
If you need extra/custom building you can hook into the builder by
extending the multimethod
qbits.spandex.sniffer-options/set-option!
source
(builder {:as options, :keys [hosts]})
(ssl-context-trust-all)
Return a SSLContext that trusts all certificate
source
(builder client sniffer options)
(encode x)
(encode-fragment value)
(chan->seq ch)
Convert a channel to a lazy sequence.
Will block on after the last element if the channel is not closed.
source
(escape-query-string query)
Escape or remove special characters in query string coming from users.
See:
https://www.elastic.co/guide/en/elasticsearch/reference/current/query-dsl-query-string-query.html#_reserved_characters
source