Class: Akasha::Storage::HttpEventStore::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/akasha/storage/http_event_store/client.rb

Overview

Eventstore HTTP client.

Constant Summary collapse

MIN_RETRY_INTERVAL =

A lower limit for a retry interval.

0
MAX_RETRY_INTERVAL =

An upper limit for a retry interval.

10.0

Instance Method Summary collapse

Constructor Details

#initialize(host: 'localhost', port: 2113, username: nil, password: nil) ⇒ Client

Creates a new client for the host and port with optional username and password for authenticating certain requests.



29
30
31
32
33
34
 
# File 'lib/akasha/storage/http_event_store/client.rb', line 29

def initialize(host: 'localhost', port: 2113, username: nil, password: nil)
  @username = username
  @password = password
  @conn = connection(host, port)
  @serializer = EventSerializer.new
end

Instance Method Details

#merge_all_by_event(name, event_names, max_retries: 0) ⇒ Object

Merges all streams into one, filtering the resulting stream so it only contains events with the specified names, using a projection.

Arguments:

`name` - name of the projection stream
`event_names` - array of event names



59
60
61
62
63
 
# File 'lib/akasha/storage/http_event_store/client.rb', line 59

def merge_all_by_event(name, event_names, max_retries: 0)
  retrying_on_network_failures(max_retries) do
    ProjectionManager.new(self).merge_all_by_event(name, event_names)
  end
end

#request(method, path, body = nil, headers = {}) ⇒ Object

Issues a generic request against the API.



79
80
81
82
83
 
# File 'lib/akasha/storage/http_event_store/client.rb', line 79

def request(method, path, body = nil, headers = {})
  body = @conn.public_send(method, path, body, auth_headers.merge(headers)).body
  return {} if body.empty?
  body
end

#retry_append_to_stream(stream_name, events, expected_version = nil, max_retries: 0) ⇒ Object

Append events to stream, idempotently retrying_on_network_failures up to ‘max_retries`



37
38
39
40
41
 
# File 'lib/akasha/storage/http_event_store/client.rb', line 37

def retry_append_to_stream(stream_name, events, expected_version = nil, max_retries: 0)
  retrying_on_network_failures(max_retries) do
    append_to_stream(stream_name, events, expected_version)
  end
end

#retry_read_events_forward(stream_name, start, count, poll = 0, max_retries: 0) ⇒ Object

Read events from stream, retrying_on_network_failures up to ‘max_retries` in case of network failures. Reads `count` events starting from `start` inclusive. Can long-poll for events if `poll` is specified.`



46
47
48
49
50
 
# File 'lib/akasha/storage/http_event_store/client.rb', line 46

def retry_read_events_forward(stream_name, start, count, poll = 0, max_retries: 0)
  retrying_on_network_failures(max_retries) do
    safe_read_events(stream_name, start, count, poll)
  end
end

#retry_read_metadata(stream_name, max_retries: 0) ⇒ Object

Reads stream metadata.



66
67
68
69
70
 
# File 'lib/akasha/storage/http_event_store/client.rb', line 66

def retry_read_metadata(stream_name, max_retries: 0)
  retrying_on_network_failures(max_retries) do
    safe_read_metadata(stream_name)
  end
end

#retry_write_metadata(stream_name, metadata) ⇒ Object

Updates stream metadata.



73
74
75
76
 
# File 'lib/akasha/storage/http_event_store/client.rb', line 73

def retry_write_metadata(stream_name, metadata)
  event = Akasha::Event.new(:stream_metadata_changed, SecureRandom.uuid, metadata)
  retry_append_to_stream("#{stream_name}/metadata", [event])
end