Module: Tusk::Observable::PG

Defined in:
lib/tusk/observable/pg.rb

Overview

An observer implementation for PostgreSQL. This module requires that your class implement a connection method that returns a database connection that this module can use.

This observer works across processes.

Example:

require 'pg'
require 'tusk/observable/pg'

class Timer
  include Tusk::Observable::PG

  def tick
    changed
    notify_observers
  end

  def connection
    Thread.current[:conn] ||= ::PG::Connection.new :dbname => 'postgres'
  end
end

class Listener
  def update
    puts "got update"
  end
end

timer = Timer.new

fork do
  timer.add_observer Listener.new
  sleep # put the process to sleep so it doesn't exit
end

loop do
  timer.tick
  sleep 1
end

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#subscribers ⇒ Object (readonly)

Returns the value of attribute subscribers.



62
63
64
 
# File 'lib/tusk/observable/pg.rb', line 62

def subscribers
  @subscribers
end

Class Method Details

.extended(klass) ⇒ Object 



50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/tusk/observable/pg.rb', line 50

def self.extended klass
  super

  klass.instance_eval do
    @sub_lock       = Mutex.new
    @observer_state = false
    @subscribers    = {}
    @_listener      = nil
    @observing      = Latch.new
  end
end

Instance Method Details

#add_observer(object, func = :update) ⇒ Object

Add observer as an observer to this object. The object will receive a notification when #changed? returns true and #notify_observers is called.

func method is called on object when notifications are sent.



115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/tusk/observable/pg.rb', line 115

def add_observer object, func = :update
  @sub_lock.synchronize do
    subscribers.fetch(channel) { |k|
      Thread.new {
        start_listener
        unwrap(connection).exec "LISTEN #{channel}"
        @observing.release
      }
      subscribers[k] = {}
    }[object] = func
  end

  @observing.await
end

#changed(state = true) ⇒ Object

Set the changed state of this object. Notifications will be sent only if the changed state is a truthy object.



96
97
98
 
# File 'lib/tusk/observable/pg.rb', line 96

def changed state = true
  @observer_state = state
end

#changed? ⇒ Boolean

Returns true if this object's state has been changed since the last call to #notify_observers.

Returns:

  • (Boolean) 


90
91
92
 
# File 'lib/tusk/observable/pg.rb', line 90

def changed?
  @observer_state
end

#count_observers ⇒ Object

Returns the number of observers associated with this object in the current process. If the object is observed across multiple processes, the returned count will not reflect the other processes.



77
78
79
 
# File 'lib/tusk/observable/pg.rb', line 77

def count_observers
  @sub_lock.synchronize { subscribers.fetch(channel, {}).length }
end

#delete_observer(o) ⇒ Object

Remove observer so that it will no longer receive notifications.



131
132
133
134
135
 
# File 'lib/tusk/observable/pg.rb', line 131

def delete_observer o
  @sub_lock.synchronize do
    subscribers.fetch(channel, {}).delete o
  end
end

#delete_observers ⇒ Object

Remove all observers associated with this object in the current process. This method will not impact observers of this object in other processes.



84
85
86
 
# File 'lib/tusk/observable/pg.rb', line 84

def delete_observers
  @sub_lock.synchronize { subscribers.delete channel }
end

#initialize(*args) ⇒ Object 



64
65
66
67
68
69
70
71
72
 
# File 'lib/tusk/observable/pg.rb', line 64

def initialize *args
  super

  @sub_lock       = Mutex.new
  @observer_state = false
  @subscribers    = {}
  @_listener      = nil
  @observing      = Latch.new
end

#notify_observers ⇒ Object

If this object's #changed? state is true, this method will notify observing objects.



102
103
104
105
106
107
108
 
# File 'lib/tusk/observable/pg.rb', line 102

def notify_observers
  return unless changed?

  unwrap(connection).exec "NOTIFY #{channel}"

  changed false
end