Class: Hivehook::Resources::StreamSinkService

Inherits:
BaseService
  • Object
show all
Defined in:
lib/hivehook/resources/stream_sink_service.rb

Constant Summary collapse

FRAGMENT = 
"id streamId name sinkType config batchSize flushInterval cursorSequence status lastFlushedAt createdAt"

Instance Method Summary collapse

Methods inherited from BaseService

#initialize

Constructor Details

This class inherits a constructor from Hivehook::Resources::BaseService

Instance Method Details

#create(input) ⇒ Object 



41
42
43
44
 
# File 'lib/hivehook/resources/stream_sink_service.rb', line 41

def create(input)
  query = "mutation($input: CreateStreamSinkInput!) { createStreamSink(input: $input) { #{FRAGMENT} } }"
  @transport.execute(query, { "input" => input })["createStreamSink"]
end

#delete(id) ⇒ Object 



51
52
53
54
 
# File 'lib/hivehook/resources/stream_sink_service.rb', line 51

def delete(id)
  query = "mutation($id: UUID!) { deleteStreamSink(id: $id) }"
  @transport.execute(query, { "id" => id })["deleteStreamSink"]
end

#get(id) ⇒ Object 



36
37
38
39
 
# File 'lib/hivehook/resources/stream_sink_service.rb', line 36

def get(id)
  query = "query($id: UUID!) { streamSink(id: $id) { #{FRAGMENT} } }"
  @transport.execute(query, { "id" => id })["streamSink"]
end

#iterate(stream_id, options = {}) ⇒ Object 



19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/hivehook/resources/stream_sink_service.rb', line 19

def iterate(stream_id, options = {})
  return enum_for(:iterate, stream_id, options) unless block_given?

  opts = options.dup
  offset = opts[:offset] || 0
  loop do
    opts[:offset] = offset
    conn = list(stream_id, opts)
    nodes = conn["nodes"] || []
    nodes.each { |node| yield node }
    page_info = conn["pageInfo"] || {}
    break if !page_info["hasNextPage"] || nodes.empty?

    offset += nodes.length
  end
end

#list(stream_id, options = {}) ⇒ Object 



8
9
10
11
12
13
14
15
16
17
 
# File 'lib/hivehook/resources/stream_sink_service.rb', line 8

def list(stream_id, options = {})
  query = "query($streamId: UUID!, $status: SinkStatus, $search: String, $limit: Int, $offset: Int, $after: String, $first: Int) {
    streamSinks(streamId: $streamId, status: $status, search: $search, limit: $limit, offset: $offset, after: $after, first: $first) {
      nodes { #{FRAGMENT} }
      pageInfo { total limit offset endCursor hasNextPage }
    }
  }"
  vars = { "streamId" => stream_id }.merge(build_variables(options, %w[status search limit offset after first]))
  @transport.execute(query, vars)["streamSinks"]
end

#update(id, input) ⇒ Object 



46
47
48
49
 
# File 'lib/hivehook/resources/stream_sink_service.rb', line 46

def update(id, input)
  query = "mutation($id: UUID!, $input: UpdateStreamSinkInput!) { updateStreamSink(id: $id, input: $input) { #{FRAGMENT} } }"
  @transport.execute(query, { "id" => id, "input" => input })["updateStreamSink"]
end