Module: DTAS::Player::ClientHandler

Included in:
DTAS::Player
Defined in:
lib/dtas/player/client_handler.rb

Overview

Copyright © 2013, Eric Wong <[email protected]> License: GPLv3 or later (www.gnu.org/licenses/gpl-3.0.txt)

Instance Method Summary collapse

Instance Method Details

#__buf_reset(buf) ⇒ Object 



302
303
304
305
306
 
# File 'lib/dtas/player/client_handler.rb', line 302

def __buf_reset(buf)
  @srv.wait_ctl(buf, :ignore)
  buf.buf_reset
  @srv.wait_ctl(buf, :wait_readable)
end

#__current_decoded_samples ⇒ Object 



185
186
187
188
189
190
 
# File 'lib/dtas/player/client_handler.rb', line 185

def __current_decoded_samples
  initial = @current.offset_samples
  decoded = @format.bytes_to_samples(bytes_decoded)
  decoded = out_samples(decoded, @format, @current.format)
  initial + decoded
end

#__current_requeue ⇒ Object 



192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
 
# File 'lib/dtas/player/client_handler.rb', line 192

def __current_requeue
  return unless @current && @current.pid

  # no need to requeue if we're already due to die
  return if @current.requeued
  @current.requeued = true

  dst = @current.dst
  # prepare to seek to the desired point based on the number of bytes which
  # passed through dst buffer we want the offset for the @current file,
  # which may have a different rate than our internal @format
  if @current.respond_to?(:infile)
    # this offset in the @current.format (not player @format)
    @queue.unshift([ @current.infile, "#{__current_decoded_samples}s" ])
  else
    # DTAS::Source::Command (hash), just rerun it
    @queue.unshift(@current.to_hsh)
  end
  # We also want to hard drop the buffer so we do not get repeated audio.
  __buf_reset(dst)
end

#__seek_offset_adj(dir, offset) ⇒ Object 



174
175
176
177
178
179
180
181
182
183
 
# File 'lib/dtas/player/client_handler.rb', line 174

def __seek_offset_adj(dir, offset)
  if offset.sub!(/s\z/, '')
    offset = offset.to_i
  else # time
    offset = @current.format.hhmmss_to_samples(offset)
  end
  n = __current_decoded_samples + (dir * offset)
  n = 0 if n < 0
  "#{n}s"
end

#__sink_activate(sink) ⇒ Object 



69
70
71
72
73
 
# File 'lib/dtas/player/client_handler.rb', line 69

def __sink_activate(sink)
  return if sink.pid
  @targets.concat(sink.spawn(@format))
  @targets.sort_by! { |t| t.sink.prio }
end

#__sink_switch(sink) ⇒ Object

called to activate/deactivate a sink



87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/dtas/player/client_handler.rb', line 87

def __sink_switch(sink)
  if sink.active
    if @current
      # maybe it's still alive for now, but it's just being killed
      # do not reactivate it until we've reaped it
      if sink.pid
        drop_sink(sink)
        sink.respawn = true
      else
        __sink_activate(sink)
      end
    end
  else
    drop_sink(sink)
  end
  # if we change any sinks, make sure the event loop watches it for
  # readability again, since we new sinks should be writable, and
  # we've stopped waiting on killed sinks
  @srv.wait_ctl(@sink_buf, :wait_readable)
end

#active_sinks ⇒ Object 



257
258
259
260
261
 
# File 'lib/dtas/player/client_handler.rb', line 257

def active_sinks
  sinks = @targets.map { |t| t.sink }
  sinks.uniq!
  sinks
end

#adjust_numeric(io, obj, k, v) ⇒ Object 



17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/dtas/player/client_handler.rb', line 17

def adjust_numeric(io, obj, k, v)
  negate = !!v.sub!(/\A-/, '')
  case v
  when %r{\A\+?\d*\.\d+\z}
    num = v.to_f
  when %r{\A\+?\d+\z}
    num = v.to_i
  else
    return io.emit("ERR #{k}=#{v} must be a float")
  end
  num = -num if negate

  if k.sub!(/\+\z/, '') # increment existing
    num += obj.__send__(k)
  elsif k.sub!(/-\z/, '') # decrement existing
    num = obj.__send__(k) - num
  # else # normal assignment
  end
  obj.__send__("#{k}=", num)
  true
end

#bytes_decoded(src = @current) ⇒ Object 



169
170
171
172
 
# File 'lib/dtas/player/client_handler.rb', line 169

def bytes_decoded(src = @current)
  bytes = src.dst.bytes_xfer - src.dst_zero_byte
  bytes = bytes < 0 ? 0 : bytes # maybe negative in case of sink errors
end

#chdir_handler(io, msg) ⇒ Object 



452
453
454
455
456
457
458
459
460
461
 
# File 'lib/dtas/player/client_handler.rb', line 452

def chdir_handler(io, msg)
  msg.size == 1 or return io.emit("ERR usage: cd DIRNAME")
  begin
    Dir.chdir(msg[0])
  rescue => e
    return io.emit("ERR chdir: #{e.message}")
  end
  # echo(%W(cd msg[0])) # should we broadcast this?
  io.emit("OK")
end

#current_expect_samples(in_samples) ⇒ Object

returns the number of samples we expect from the source this takes into account sample rate differences between the source and internal player format



224
225
226
 
# File 'lib/dtas/player/client_handler.rb', line 224

def current_expect_samples(in_samples) # @current.samples
  out_samples(in_samples, @current.format, @format)
end

#current_handler(io, msg) ⇒ Object

show current info about what’s playing returns non-blocking iterator retval



265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
 
# File 'lib/dtas/player/client_handler.rb', line 265

def current_handler(io, msg)
  tmp = {}
  if @current
    tmp["current"] = s = @current.to_hsh
    s["spawn_at"] = @current.spawn_at
    s["pid"] = @current.pid

    # this offset and samples in the player @format (not @current.format)
    decoded = @format.bytes_to_samples(bytes_decoded)
    if @current.respond_to?(:infile)
      initial = tmp["current_initial"] = @current.offset_samples
      initial = out_samples(initial, @current.format, @format)
      tmp["current_expect"] = current_expect_samples(s["samples"])
      s["format"] = @current.format.to_hash.delete_if { |_,v| v.nil? }
    else
      initial = 0
      tmp["current_expect"] = nil
      s["format"] = @format.to_hash.delete_if { |_,v| v.nil? }
    end
    tmp["current_offset"] = initial + decoded
  end
  tmp["current_inflight"] = @sink_buf.inflight
  tmp["format"] = @format.to_hash.delete_if { |_,v| v.nil? }
  tmp["paused"] = @paused
  rg = @rg.to_hsh
  tmp["rg"] = rg unless rg.empty?
  if @targets[0]
    sinks = active_sinks
    tmp["sinks"] = sinks.map! do |sink|
      h = sink.to_hsh
      h["pid"] = sink.pid
      h
    end
  end
  io.emit(tmp.to_yaml)
end

#do_pause ⇒ Object 



325
326
327
328
329
330
 
# File 'lib/dtas/player/client_handler.rb', line 325

def do_pause
  return if @paused
  echo("pause")
  @paused = true
  __current_requeue
end

#do_play ⇒ Object 



332
333
334
335
336
337
 
# File 'lib/dtas/player/client_handler.rb', line 332

def do_play
  # no echo, next_source will echo on new track
  @paused = false
  return if @current && @current.pid
  next_source(@queue.shift)
end

#do_play_pause ⇒ Object 



339
340
341
 
# File 'lib/dtas/player/client_handler.rb', line 339

def do_play_pause
  @paused ? do_play : do_pause
end

#do_seek(io, offset) ⇒ Object 



343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
 
# File 'lib/dtas/player/client_handler.rb', line 343

def do_seek(io, offset)
  if @current && @current.pid
    if @current.respond_to?(:infile)
      begin
        if offset.sub!(/\A\+/, '')
          offset = __seek_offset_adj(1, offset)
        elsif offset.sub!(/\A-/, '')
          offset = __seek_offset_adj(-1, offset)
        # else: pass to sox directly
        end
      rescue ArgumentError
        return io.emit("ERR bad time format")
      end
      @queue.unshift([ @current.infile, offset ])
      __buf_reset(@current.dst) # trigger EPIPE
    else
      return io.emit("ERR unseekable")
    end
  elsif @paused
    case file = @queue[0]
    when String
      @queue[0] = [ file, offset ]
    when Array
      file[1] = offset
    else
      return io.emit("ERR unseekable")
    end
  # unpaused case... what do we do?
  end
  io.emit("OK")
end

#drop_sink(sink) ⇒ Object 



75
76
77
78
79
80
81
82
83
84
 
# File 'lib/dtas/player/client_handler.rb', line 75

def drop_sink(sink)
  @targets.delete_if do |t|
    if t.sink == sink
      drop_target(t)
      true
    else
      false
    end
  end
end

#env_handler(io, msg) ⇒ Object 



409
410
411
412
413
414
415
416
417
418
419
420
421
 
# File 'lib/dtas/player/client_handler.rb', line 409

def env_handler(io, msg)
  msg.each do |kv|
    case kv
    when %r{\A([^=]+)=(.*)\z}
      ENV[$1] = $2
    when %r{\A([^=]+)#}
      ENV.delete($1)
    else
      return io.emit("ERR bad env")
    end
  end
  io.emit("OK")
end

#format_handler(io, msg) ⇒ Object 



382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
 
# File 'lib/dtas/player/client_handler.rb', line 382

def format_handler(io, msg)
  new_fmt = @format.dup
  msg.each do |kv|
    k, v = kv.split(/=/, 2)
    case k
    when "type"
      new_fmt.valid_type?(v) or return io.emit("ERR invalid file type")
      new_fmt.type = v
    when "channels", "bits", "rate"
      rv = set_uint(io, kv, v, false) { |u| new_fmt.__send__("#{k}=", u) }
      rv == true or return rv
    when "endian"
      new_fmt.valid_endian?(v) or return io.emit("ERR invalid endian")
      new_fmt.endian = v
    end
  end

  if new_fmt != @format
    restart_pipeline # calls __current_requeue

    # we must assign this after __current_requeue since __current_requeue
    # relies on the old @format for calculation
    @format = new_fmt
  end
  io.emit("OK")
end

#out_samples(in_samples, infmt, outfmt) ⇒ Object 



214
215
216
217
218
219
 
# File 'lib/dtas/player/client_handler.rb', line 214

def out_samples(in_samples, infmt, outfmt)
  in_rate = infmt.rate
  out_rate = outfmt.rate
  return in_samples if in_rate == out_rate # easy!
  (in_samples * out_rate / in_rate.to_f).round
end

#play_pause_handler(io, command) ⇒ Object 



314
315
316
317
318
319
320
321
322
323
 
# File 'lib/dtas/player/client_handler.rb', line 314

def play_pause_handler(io, command)
  prev = @paused
  __send__("do_#{command}")
  io.emit({
    "paused" => {
      "before" => prev,
      "after" => @paused,
    }
  }.to_yaml)
end

#restart_pipeline ⇒ Object 



375
376
377
378
379
380
 
# File 'lib/dtas/player/client_handler.rb', line 375

def restart_pipeline
  return if @paused
  __current_requeue
  @sinks.each_value { |sink| sink.respawn = sink.active }
  @targets.each { |t| drop_target(t) }.clear
end

#rg_handler(io, msg) ⇒ Object 



228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/dtas/player/client_handler.rb', line 228

def rg_handler(io, msg)
  return io.emit(@rg.to_hsh.to_yaml) if msg.empty?
  before = @rg.to_hsh
  msg.each do |kv|
    k, v = kv.split(/=/, 2)
    case k
    when "mode"
      case v
      when "off"
        @rg.mode = nil
      else
        DTAS::RGState::RG_MODE.include?(v) or
          return io.emit("ERR rg mode invalid")
        @rg.mode = v
      end
    when "fallback_track"
      rv = set_bool(io, kv, v) { |b| @rg.fallback_track = b }
      rv == true or return rv
    when %r{(?:gain_threshold|norm_threshold|
            preamp|norm_level|fallback_gain)[+-]?\z}x
      rv = adjust_numeric(io, @rg, k, v)
      rv == true or return rv
    end
  end
  after = @rg.to_hsh
  __current_requeue if before != after
  io.emit("OK")
end

#set_bool(io, kv, v) ⇒ Object

returns true on success, wait_ctl arg on error



7
8
9
10
11
12
13
14
15
 
# File 'lib/dtas/player/client_handler.rb', line 7

def set_bool(io, kv, v)
  case v
  when "false" then yield(false)
  when "true" then yield(true)
  else
    return io.emit("ERR #{kv} must be true or false")
  end
  true
end

#set_int(io, kv, v, null_ok) ⇒ Object

returns true on success, wait_ctl arg on error



40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/dtas/player/client_handler.rb', line 40

def set_int(io, kv, v, null_ok)
  case v
  when %r{\A-?\d+\z}
    yield(v.to_i)
  when ""
    null_ok or return io.emit("ERR #{kv} must be defined")
    yield(nil)
  else
    return io.emit("ERR #{kv} must an integer")
  end
  true
end

#set_uint(io, kv, v, null_ok) ⇒ Object

returns true on success, wait_ctl arg on error



54
55
56
57
58
59
60
61
62
63
64
65
66
67
 
# File 'lib/dtas/player/client_handler.rb', line 54

def set_uint(io, kv, v, null_ok)
  case v
  when %r{\A\d+\z}
    yield(v.to_i)
  when %r{\A0x[0-9a-fA-F]+\z}i # hex
    yield(v.to_i(16))
  when ""
    null_ok or return io.emit("ERR #{kv} must be defined")
    yield(nil)
  else
    return io.emit("ERR #{kv} must an non-negative integer")
  end
  true
end

#sink_handler(io, msg) ⇒ Object

returns a wait_ctl arg



109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
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
157
158
159
160
161
162
163
164
165
166
167
 
# File 'lib/dtas/player/client_handler.rb', line 109

def sink_handler(io, msg)
  name = msg[1]
  case msg[0]
  when "ls"
    io.emit(Shellwords.join(@sinks.keys.sort))
  when "rm"
    sink = @sinks.delete(name) or return io.emit("ERR #{name} not found")
    drop_sink(sink)
    io.emit("OK")
  when "ed"
    sink = @sinks[name] || (new_sink = DTAS::Sink.new)

    # allow things that look like audio device names ("hw:1,0" , "/dev/dsp")
    # or variable names.
    sink.valid_name?(name) or return io.emit("ERR sink name invalid")

    sink.name = name
    active_before = sink.active

    # multiple changes may be made at once
    msg[2..-1].each do |kv|
      k, v = kv.split(/=/, 2)
      case k
      when %r{\Aenv\.([^=]+)\z}
        sink.env[$1] = v
      when %r{\Aenv#([^=]+)\z}
        v == nil or return io.emit("ERR unset env has no value")
        sink.env.delete($1)
      when "prio"
        rv = set_int(io, kv, v, false) { |i| sink.prio = i }
        rv == true or return rv
        @targets.sort_by! { |t| t.sink.prio } if sink.active
      when "nonblock", "active"
        rv = set_bool(io, kv, v) { |b| sink.__send__("#{k}=", b) }
        rv == true or return rv
      when "pipe_size"
        rv = set_uint(io, kv, v, true) { |u| sink.__send__("#{k}=", u) }
        rv == true or return rv
      when "command" # nothing to validate, this could be "rm -rf /" :>
        sink.command = v.empty? ? DTAS::Sink::SINK_DEFAULTS["command"] : v
      end
    end

    @sinks[name] = new_sink if new_sink # no errors? it's a new sink!

    # start or stop a sink if its active= flag changed.  Additionally,
    # account for a crashed-but-marked-active sink.  The user may have
    # fixed the command to not crash it.
    if (active_before != sink.active) || (sink.active && !sink.pid)
      __sink_switch(sink)
    end
    io.emit("OK")
  when "cat"
    sink = @sinks[name] or return io.emit("ERR #{name} not found")
    io.emit(sink.to_hsh.to_yaml)
  else
    io.emit("ERR unknown sink op #{msg[0]}")
  end
end

#skip_handler(io, msg) ⇒ Object 



308
309
310
311
312
 
# File 'lib/dtas/player/client_handler.rb', line 308

def skip_handler(io, msg)
  __current_drop
  echo("skip")
  io.emit("OK")
end

#source_handler(io, msg) ⇒ Object 



423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
 
# File 'lib/dtas/player/client_handler.rb', line 423

def source_handler(io, msg)
  case msg.shift
  when "cat"
    io.emit({
      "command" => @srccmd || DTAS::Source::SOURCE_DEFAULTS["command"],
      "env" => @srcenv,
    }.to_yaml)
  when "ed"
    before = [ @srccmd, @srcenv ].inspect
    msg.each do |kv|
      k, v = kv.split(/=/, 2)
      case k
      when "command"
        @srccmd = v.empty? ? nil : v
      when %r{\Aenv\.([^=]+)\z}
        @srcenv[$1] = v
      when %r{\Aenv#([^=]+)\z}
        v == nil or return io.emit("ERR unset env has no value")
        @srcenv.delete($1)
      end
    end
    after = [ @srccmd, @srcenv ].inspect
    __current_requeue if before != after
    io.emit("OK")
  else
    io.emit("ERR unknown source op")
  end
end