Class: OmfCommon::Comm::Local::Communicator

Inherits:
OmfCommon::Comm show all
Defined in:
omf_common/lib/omf_common/comm/local/local_communicator.rb

Instance Method Summary (collapse)

Methods inherited from OmfCommon::Comm

#conn_info, init, #initialize, instance, #local_address, #local_topic, #on_interrupted, #options, #string_to_topic_address, #subscribe

Constructor Details

This class inherits a constructor from OmfCommon::Comm

Instance Method Details

- (Object) broadcast_file(file_path, topic_url = nil, opts = {}, &block) 



58
59
60
61
62
 
# File 'omf_common/lib/omf_common/comm/local/local_communicator.rb', line 58

def broadcast_file(file_path, topic_url = nil, opts = {}, &block)
  topic_url ||= SecureRandom.uuid
  @distributed_files[topic_url] = file_path
  "bdcst:local:#{topic_url}"
end

- (Object) create_topic(topic, &block)

Create a new pubsub topic with additional configuration

Parameters:

  • topic (String) —

    Pubsub topic name



31
32
33
34
35
36
37
 
# File 'omf_common/lib/omf_common/comm/local/local_communicator.rb', line 31

def create_topic(topic, &block)
  t = OmfCommon::Comm::Local::Topic.create(topic)
  if block
    block.call(t)
  end
  t
end

- (Object) delete_topic(topic, &block)

Delete a pubsub topic

Parameters:

  • topic (String) —

    Pubsub topic name



42
43
44
45
46
47
48
 
# File 'omf_common/lib/omf_common/comm/local/local_communicator.rb', line 42

def delete_topic(topic, &block)
  if t = OmfCommon::CommProvider::Local::Topic.find(topic)
    t.release
  else
    warn "Attempt to delete unknown topic '#{topic}"
  end        
end

- (Object) disconnect(opts = {})

Shut down comms layer



25
26
 
# File 'omf_common/lib/omf_common/comm/local/local_communicator.rb', line 25

def disconnect(opts = {})
end

- (Object) init(opts = {})

Initialize comms layer



19
20
21
22
 
# File 'omf_common/lib/omf_common/comm/local/local_communicator.rb', line 19

def init(opts = {})
  @distributed_files = {}
					super
end

- (Object) on_connected(&block) 



50
51
52
53
54
55
56
 
# File 'omf_common/lib/omf_common/comm/local/local_communicator.rb', line 50

def on_connected(&block)
  return unless block
  
  OmfCommon.eventloop.after(0) do
    block.arity == 1 ? block.call(self) : block.call
  end
end

- (Object) receive_file(topic_url, file_path = nil, opts = {}, &block) 



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'omf_common/lib/omf_common/comm/local/local_communicator.rb', line 64

def receive_file(topic_url, file_path = nil, opts = {}, &block)
  if topic_url.start_with? 'local:'
    topic_url = topic_url[6 .. -1]
  end
  file_path ||= File.join(Dir.tmpdir, Dir::Tmpname.make_tmpname('bdcast', ''))
  OmfCommon.eventloop.after(0) do
    #puts ">>>>>> #{topic_url}::#{@distributed_files.keys}"
    unless original = @distributed_files[topic_url]
      raise "File '#{topic_url}' hasn't started broadcasting"
    end
    mime_type = `file -b --mime-type #{original}`
    `cp #{original} #{file_path}`
    unless $?.success?
      error "Couldn't copy '#{original}' to '#{file_path}'"
    end
    if block
      block.call({action: :done, mime_type: mime_type.strip, path: file_path, size: -1, received: -1})
    end
  end
  file_path
end