Class: ThreeScale::Backend::StorageAsync::Pipeline

Inherits:
Object
  • Object
show all
Defined in:
lib/3scale/backend/storage_async/pipeline.rb

Overview

This class accumulates commands and sends several of them in a single request, instead of sending them one by one.

Defined Under Namespace

Classes: PipelineSharedBetweenFibers

Constant Summary collapse

Error = 
Class.new StandardError

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Pipeline

Returns a new instance of Pipeline.



28
29
30
31
32
33
34
35
36
37
38
 
# File 'lib/3scale/backend/storage_async/pipeline.rb', line 28

def initialize
  # Each command is an array where the first element is the name of the
  # command ('SET', 'GET', etc.) and the rest of elements are the
  # parameters for that command.
  # Ex: ['SET', 'some_key', 42].
  @commands = []

  # Save the ID of the fiber that created the Pipeline so later we
  # can check that this pipeline is not shared between fibers.
  @fiber_id = Fiber.current.object_id
end

Instance Method Details

#call(*args) ⇒ Object

In the async-redis lib, all the commands are run with .call: client.call(‘GET’, ‘a’), client.call(‘SET’, ‘b’, ‘1’), etc. This method just accumulates the commands and their params.



43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/3scale/backend/storage_async/pipeline.rb', line 43

def call(*args)
  if @fiber_id != Fiber.current.object_id
    raise PipelineSharedBetweenFibers
  end

  @commands << args

  # Some Redis commands in StorageAsync compare the result with 0.
  # For example, EXISTS. We return an integer so the comparison does
  # not raise an error. It does not matter which int, because here we
  # only care about adding the command to @commands.

  1
end

#run(redis_async_client) ⇒ Object

Send to redis all the accumulated commands. Returns an array with the result for each command in the same order that they added with .call().



61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
 
# File 'lib/3scale/backend/storage_async/pipeline.rb', line 61

def run(redis_async_client)
  responses = redis_async_client.call_pipeline(@commands)

  responses.zip(@commands).map do |resp, cmd|
    command_name = cmd.first.to_s.upcase

    if CHECK_EQUALS_ONE.include?(command_name)
      resp.to_i == 1
    elsif CHECK_GREATER_THAN_0.include?(command_name)
      resp.to_i > 0
    else
      resp
    end
  end
end