Module: Parse::Console

Defined in:
lib/parse/console.rb

Constant Summary collapse

DEFAULT_WATCH_EVENTS = 
[:create, :update, :delete, :enter, :leave].freeze
DEFAULT_WAIT_FOR_EVENTS = 
[:create, :enter].freeze

Class Method Summary collapse

Class Method Details

.wait_for(klass, where: {}, on: nil, timeout: nil, fields: nil, session_token: nil) {|object| ... } ⇒ Parse::Object

Block until a row matching the predicate arrives via LiveQuery, then return that row. Useful for wait until the job flips, wait until the inbox row lands, integration tests, etc.

By default the first :create/:enter event resolves the wait. Pass a block to require the event also satisfy a predicate — wait_for(Job, where: { kind: "import" }) { |j| j.status == "done" } will keep waiting until both the query matches AND the block returns truthy.

Examples:

Wait for the first row of a class

first = Parse.wait_for(Notification)

Wait with a query and a predicate

done = Parse.wait_for(Job, where: { kind: "import" },
                       timeout: 60) { |j| j.status == "done" }

Parameters:

  • klass (Class) —

    a Parse::Object subclass.

  • where (Hash) (defaults to: {}) —

    optional query constraints applied server-side.

  • on (Symbol, Array<Symbol>) (defaults to: nil) —

    which event types to count (default: [:create, :enter]; pass :update for status-flip watching).

  • timeout (Numeric, nil) (defaults to: nil) —

    seconds to wait. nil = forever.

  • fields (Array<String>, nil) (defaults to: nil) —

    field filter for update events.

  • session_token (String, nil) (defaults to: nil) —

    explicit override; defaults to Parse.current_session_token.

Yields:

  • (object) —

    optional predicate; must return truthy to resolve.

Returns:

Raises:

  • (Timeout::Error) —

    when timeout: elapses with no match.



128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/parse/console.rb', line 128

def wait_for(klass, where: {}, on: nil, timeout: nil, fields: nil,
                    session_token: nil, &predicate)
  events = Array(on || DEFAULT_WAIT_FOR_EVENTS).map(&:to_sym)
  queue = Queue.new
  sub = _open_subscription(klass, where: where, fields: fields, session_token: session_token)

  events.each do |ev|
    sub.on(ev) do |obj, _original = nil|
      begin
        next if predicate && !predicate.call(obj)
        queue << obj
      rescue StandardError => e
        queue << e
      end
    end
  end
  sub.on(:error) { |err| queue << err }

  result = if timeout
      Timeout.timeout(timeout) { queue.pop }
    else
      queue.pop
    end

  raise result if result.is_a?(Exception)
  result
ensure
  sub.unsubscribe if sub && sub.respond_to?(:unsubscribe)
end

.watch(klass, where: {}, on: nil, fields: nil, session_token: nil) {|event, object| ... } ⇒ Integer

Open a LiveQuery subscription on klass and block until SIGINT (Ctrl-C). Intended for REPL / console use — emits arriving events to $stdout by default, or yields each one to a caller-supplied block.

Examples:

Tail every event for a class

Parse.watch(Post)
# ^C to stop

Tail with a query and a custom handler

Parse.watch(Post, where: { category: "alerts" }) do |event, obj|
  puts "[#{event}] #{obj.title} (#{obj.id})"
end

Admin-style with master key (no ambient session)

Parse.with_session(nil) { Parse.watch(JobRun) }

Parameters:

  • klass (Class) —

    a Parse::Object subclass.

  • where (Hash) (defaults to: {}) —

    optional query constraints.

  • on (Symbol, Array<Symbol>, nil) (defaults to: nil) —

    which event types to subscribe to (default: all of :create/:update/:delete/:enter/:leave).

  • fields (Array<String>, nil) (defaults to: nil) —

    only fire updates when these fields change.

  • session_token (String, nil) (defaults to: nil) —

    explicit override; defaults to Parse.current_session_token.

Yields:

  • (event, object) —

    called for each emitted event when a block is supplied. event is one of the watched symbols; object is the row.

Returns:

  • (Integer) —

    the count of events delivered before the caller interrupted (Ctrl-C) or the subscription was torn down.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
 
# File 'lib/parse/console.rb', line 65

def watch(klass, where: {}, on: nil, fields: nil, session_token: nil, &block)
  events = Array(on || DEFAULT_WATCH_EVENTS).map(&:to_sym)
  printer = block_given? ? block : ->(ev, obj) {
    title = obj.respond_to?(:id) ? obj.id : obj.inspect
    # The row is tenant data arriving over a live-query socket and this
    # line goes straight to the operator's terminal, so escape it.
    puts "[#{Time.now.iso8601}] #{klass.parse_class}.#{ev} " \
         "#{Parse::TerminalSafe.sanitize_line(title)}"
  }

  delivered = 0
  counter_lock = Monitor.new
  sub = _open_subscription(klass, where: where, fields: fields, session_token: session_token)

  events.each do |ev|
    sub.on(ev) do |obj, _original = nil|
      counter_lock.synchronize { delivered += 1 }
      begin
        printer.call(ev, obj)
      rescue StandardError => e
        # The message can quote the row that triggered it.
        warn "[Parse.watch] handler raised #{e.class}: " \
             "#{Parse::TerminalSafe.sanitize_line(e.message)}"
      end
    end
  end
  sub.on(:error) { |err| warn "[Parse.watch] error: #{Parse::TerminalSafe.sanitize_line(err)}" }

  _block_until_interrupt
  delivered
ensure
  sub.unsubscribe if sub && sub.respond_to?(:unsubscribe)
end