Class: Tupelo::Archiver

Inherits:
Client show all
Includes:
Funl::HistoryClient
Defined in:
lib/tupelo/archiver.rb,
lib/tupelo/archiver.rb,
lib/tupelo/archiver/worker.rb,
lib/tupelo/archiver/persister.rb,
lib/tupelo/archiver/tuplespace.rb,
lib/tupelo/archiver/persistent-tuplespace.rb

Overview

should manipulate tuples as strings (at least in msgpack/json cases) instead of objects – use msgpack extension for #hash and #== on packed objects

Defined Under Namespace

Classes: ForkRequest, PersistentTuplespace, Persister, Tuplespace, Worker

Constant Summary collapse

ZERO_TOLERANCE =

How many tuples with count=0 do we permit before cleaning up?

1000

Constants inherited from Client

Client::TUPELO_META_KEY, Client::TUPELO_SUBSPACE_TAG

Instance Attribute Summary collapse

Attributes inherited from Client

#tuplespace, #worker

Instance Method Summary collapse

Methods inherited from Client

#inspect, #log, #make_queue, #make_scheduler, #or, #trace_loop, #tracing?

Methods included from Client::Api

#abort, #define_subspace, #notifier, #pulse_nowait, #pulse_wait, #read_all, #read_nowait, #read_wait, #subspace, #take, #take_nowait, #trans_class, #transaction, #use_subspaces!, #write_nowait, #write_wait

Constructor Details

#initialize(server, tuplespace: Tupelo::Archiver::Tuplespace, persist_dir: nil, **opts) ⇒ Archiver

Returns a new instance of Archiver.



25
26
27
28
29
30
31
 
# File 'lib/tupelo/archiver.rb', line 25

def initialize server,
    tuplespace: Tupelo::Archiver::Tuplespace,
    persist_dir: nil, **opts
  @server = server
  @persist_dir = persist_dir
  super arc: nil, tuplespace: tuplespace, **opts
end

Instance Attribute Details

#persist_dir ⇒ Object (readonly)

Returns the value of attribute persist_dir.



19
20
21
 
# File 'lib/tupelo/archiver.rb', line 19

def persist_dir
  @persist_dir
end

#server ⇒ Object (readonly)

Returns the value of attribute server.



18
19
20
 
# File 'lib/tupelo/archiver.rb', line 18

def server
  @server
end

#server_thread ⇒ Object (readonly)

Returns the value of attribute server_thread.



20
21
22
 
# File 'lib/tupelo/archiver.rb', line 20

def server_thread
  @server_thread
end

Instance Method Details

#handle_conn(conn) ⇒ Object 



86
87
88
89
90
91
92
93
94
 
# File 'lib/tupelo/archiver.rb', line 86

def handle_conn conn
  log.debug {"accepted #{conn.inspect}"}
  begin
    worker << ForkRequest.new(conn)
  rescue => ex
    log.error ex
    raise
  end
end

#make_worker ⇒ Object 



54
55
56
57
58
59
60
 
# File 'lib/tupelo/archiver.rb', line 54

def make_worker
  if persist_dir ## ???
    Tupelo::Archiver::Worker.new self, persist_dir: persist_dir
  else
    Tupelo::Archiver::Worker.new self
  end
end

#run ⇒ Object 



74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/tupelo/archiver.rb', line 74

def run
  loop do
    ## nonblock_accept?
    Thread.new(server.accept) do |conn|
      handle_conn conn
    end
  end
rescue => ex
  log.error ex
  raise
end

#start ⇒ Object 



62
63
64
65
66
67
 
# File 'lib/tupelo/archiver.rb', line 62

def start
  super # start worker thread
  @server_thread = Thread.new do
    run
  end
end

#stop ⇒ Object 



69
70
71
72
 
# File 'lib/tupelo/archiver.rb', line 69

def stop
  server_thread.kill if server_thread
  super # stop worker thread
end