Class: IChannel::Redis

Inherits:
Object
  • Object
show all
Defined in:
lib/ichannel/redis.rb

Instance Method Summary collapse

Constructor Details

#initialize(serializer, options) ⇒ IChannel::Redis

Parameters:

  • serializer (#dump, #load) —

    A serializer.

  • options (Hash) —

    A Hash of options to pass onto the redis-rb client.



12
13
14
15
16
17
18
 
# File 'lib/ichannel/redis.rb', line 12

def initialize(serializer, options)
  @serializer = serializer
  @key = options.delete(:key) || "channel"
  @redis = ::Redis.new(options)
  @last_msg = nil
  @closed = false
end

Instance Method Details

#close ⇒ Boolean

Close the channel.

Returns:

  • (Boolean) —

    Returns true when the channel has been closed.



34
35
36
37
38
39
40
 
# File 'lib/ichannel/redis.rb', line 34

def close
  unless closed?
    @redis.quit
    @last_msg = nil
    @closed = true
  end
end

#closed? ⇒ Boolean

Returns true when the channel is closed.

Returns:

  • (Boolean) —

    Returns true when the channel is closed.



24
25
26
 
# File 'lib/ichannel/redis.rb', line 24

def closed?
  @closed
end

#empty? ⇒ Boolean

Returns true when the channel is empty.

Returns:

  • (Boolean) —

    Returns true when the channel is empty.



142
143
144
 
# File 'lib/ichannel/redis.rb', line 142

def empty?
  @redis.llen(@key) == 0
end

#last_msg ⇒ Object

Returns the last message written to the channel.

Returns:

  • (Object) —

    Returns the last message written to the channel.



131
132
133
134
135
136
 
# File 'lib/ichannel/redis.rb', line 131

def last_msg
  while readable?
    @last_msg = get
  end
  @last_msg
end

#readable? ⇒ Boolean

Returns true when the channel is readable.

Returns:

  • (Boolean) —

    Returns true when the channel is readable.



150
151
152
 
# File 'lib/ichannel/redis.rb', line 150

def readable?
  !closed? && !empty?
end

#recv ⇒ Object Also known as: get

Read an object from the channel.

Returns:

  • (Object) —

    The object read from the channel.

Raises:

  • (IOError) —

    When the channel is closed or empty.



96
97
98
 
# File 'lib/ichannel/redis.rb', line 96

def recv
  recv! nil
end

#recv!(timeout = 1) ⇒ Object Also known as: get!

Returns The object read from the channel.

Parameters:

  • timeout (Fixnum) (defaults to: 1) —

    The number of seconds (as an Integer) to wait before a time out will occur.

Returns:

  • (Object) —

    The object read from the channel.

Raises:

  • (IOError) —

    When the channel is closed or empty.

  • (Timeout::Error) —

    When timeout seconds pass and the read has not completed.



114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/ichannel/redis.rb', line 114

def recv!(timeout = 1)
  if closed?
    raise IOError, 'The channel cannot be read from (closed).'
  end
  _, payload = @redis.brpop @key, timeout
  if payload
    @last_msg = @serializer.load(payload)
  else
    raise Timeout::Error, "timeout on read after #{timeout} seconds"
  end
end

#write(object) ⇒ void Also known as: put

This method returns an undefined value.

Add an object to the channel.

Parameters:

  • object (Object) —

    The object to add to the channel.

Raises:

  • (IOError) —

    When the channel is closed.



54
55
56
 
# File 'lib/ichannel/redis.rb', line 54

def write(object)
  write!(object, nil)
end

#write!(object, timeout = 0.1) ⇒ void Also known as: put!

This method returns an undefined value.

Add an object to the channel.

Parameters:

  • object (Object) —

    The object to add to the channel.

  • timeout (Fixnum) (defaults to: 0.1) —

    The amount of time to wait for the write to complete.

Raises:

  • (IOError) —

    When the channel is closed.

  • (Timeout::Error) —

    When the write does not complete in time.



76
77
78
79
80
81
82
83
84
 
# File 'lib/ichannel/redis.rb', line 76

def write!(object, timeout = 0.1)
  if closed?
    raise IOError, 'The channel cannot be written to (closed)'
  end
  Timeout.timeout(timeout) do
    dump = @serializer.dump object
    @redis.lpush @key, dump
  end
end