19916213389

require 'thread'
require 'atomic'

class EditMediator

  # The table of all editing clients
  @active_edits = Atomic.new({})

  # Queue used to serialise access to the active edit table
  @queue = Queue.new

  def self.record(user, uri, lock_version, last_report_time)
    # Check whether the lock version is out of date, or whether someone else is
    # editing.
    status = @active_edits.value[uri]

    if status && status['lock_version'] && status['lock_version'] > lock_version
      # Stale
      return {:status => "stale"}
    end


    # Record the fact that this user is editing
    @queue << {
      :type => :update,
      :values => [user, uri, last_report_time.iso8601]
    }

    if status && status['edited_by'].keys.count > 1
      # Someone else is editing too!
      {
        :status => "opened_for_editing",
        'edited_by' => Hash[status['edited_by'].reject {|u, _| u == user}]
      }
    else
      {:status => "ok"}
    end
  end


  class UpdateThread

    # We send our local copy of the active edits table to the backend
    # periodically.  Set this the same as the client's INTERVAL_PERIOD so we
    # don't add too much delay to a client's update hitting the backend.
    SYNC_WITH_BACKEND_SECONDS = 10

    def initialize(active_edits, queue)
      @active_edits = active_edits
      @queue = queue
    end


    def record_update_locally(user, uri, last_report_time)
      @active_edits.update {|edits|
        entry = edits[uri] ? edits[uri].clone : {'edited_by' => {}}
        entry['edited_by'][user] = last_report_time

        edits.merge(uri => entry)
      }
    end


    def log_in_to_backend(force = false)
      @backend_session = nil if force

      if !@backend_session
        response = JSONModel::HTTP.post_form("/users/#{AppConfig[:staff_username]}/login",
                                             {
                                               "expiring" => "false",
                                               "password" => AppConfig[:staff_user_secret]
                                             })

        auth = ASUtils.json_parse(response.body)

        @backend_session = auth['session']
      end

      @backend_session
    end


    def synchronise_with_backend
      snapshot = @active_edits.value
      edits = JSONModel(:active_edits).new

      edits.active_edits = snapshot.map {|uri, entries|
        entries['edited_by'].map {|user, last_time|
          {
            'uri' => uri,
            'user' => user,
            'time' => last_time,
          }
        }
      }.flatten(1)

      JSONModel::HTTP.current_backend_session = log_in_to_backend

      begin
        updated = edits.save({}, true)
        @active_edits.update {|old_val| updated}
      rescue AccessDeniedException
        log_in_to_backend(true)
      end
    end

    def start
      # The main thread: respond to updates and manage the local editing state.
      Thread.new do
        while true
          begin
            request = @queue.pop

            if request[:type] == :update
              self.record_update_locally(*request[:values])
            elsif request[:type] == :sync
              self.synchronise_with_backend
            end
          rescue
            $stderr.puts("ERROR: Edit mediator: #{$!}: #{$@}")
            sleep 5
          end
        end
      end


      # A separate thread to trigger a sync with the backend every now and then.
      Thread.new do
        while true
          sleep SYNC_WITH_BACKEND_SECONDS
          @queue << {:type => :sync}
        end
      end
    end
  end


  # When the system starts, run the update thread
  UpdateThread.new(@active_edits, @queue).start
end

1	require 'thread'	11✔
2	require 'atomic'	11✔
3
4	class EditMediator	11✔
5
6	# The table of all editing clients
7	@active_edits = Atomic.new({})	11✔
8
9	# Queue used to serialise access to the active edit table
10	@queue = Queue.new	11✔
11
12	def self.record(user, uri, lock_version, last_report_time)	11✔
13	# Check whether the lock version is out of date, or whether someone else is
14	# editing.
15	status = @active_edits.value[uri]	747✔
16
17	if status && status['lock_version'] && status['lock_version'] > lock_version	747✔
18	# Stale
19	return {:status => "stale"}	×
20	end
21
22
23	# Record the fact that this user is editing
24	@queue << {	1,494✔
25	:type => :update,
26	:values => [user, uri, last_report_time.iso8601]
27	}
28
29	if status && status['edited_by'].keys.count > 1	747✔
30	# Someone else is editing too!
31	{	×
32	:status => "opened_for_editing",
33	'edited_by' => Hash[status['edited_by'].reject {\|u, _\| u == user}]	×
34	}
35	else
36	{:status => "ok"}	747✔
37	end
38	end
39
40
41	class UpdateThread	11✔
42
43	# We send our local copy of the active edits table to the backend
44	# periodically. Set this the same as the client's INTERVAL_PERIOD so we
45	# don't add too much delay to a client's update hitting the backend.
46	SYNC_WITH_BACKEND_SECONDS = 10	11✔
47
48	def initialize(active_edits, queue)	11✔
49	@active_edits = active_edits	11✔
50	@queue = queue	11✔
51	end
52
53
54	def record_update_locally(user, uri, last_report_time)	11✔
55	@active_edits.update {\|edits\|	747✔
56	entry = edits[uri] ? edits[uri].clone : {'edited_by' => {}}	747✔
57	entry['edited_by'][user] = last_report_time	747✔
58
59	edits.merge(uri => entry)	747✔
60	}
61	end
62
63
64	def log_in_to_backend(force = false)	11✔
65	@backend_session = nil if force	771✔
66
67	if !@backend_session	771✔
68	response = JSONModel::HTTP.post_form("/users/#{AppConfig[:staff_username]}/login",	33✔
69	{
70	"expiring" => "false",
71	"password" => AppConfig[:staff_user_secret]
72	})
73
74	auth = ASUtils.json_parse(response.body)	11✔
75
76	@backend_session = auth['session']	11✔
77	end
78
79	@backend_session	749✔
80	end
81
82
83	def synchronise_with_backend	11✔
84	snapshot = @active_edits.value	771✔
85	edits = JSONModel(:active_edits).new	771✔
86
87	edits.active_edits = snapshot.map {\|uri, entries\|	771✔
88	entries['edited_by'].map {\|user, last_time\|	1,495✔
89	{	×
90	'uri' => uri,	1,492✔
91	'user' => user,
92	'time' => last_time,
93	}
94	}
95	}.flatten(1)
96
97	JSONModel::HTTP.current_backend_session = log_in_to_backend	771✔
98
99	begin	738✔
100	updated = edits.save({}, true)	749✔
101	@active_edits.update {\|old_val\| updated}	1,496✔
102	rescue AccessDeniedException
103	log_in_to_backend(true)	×
104	end
105	end
106
107	def start	11✔
108	# The main thread: respond to updates and manage the local editing state.
109	Thread.new do	11✔
110	while true	11✔
111	begin	1,517✔
112	request = @queue.pop	1,528✔
113
114	if request[:type] == :update	1,518✔
115	self.record_update_locally(*request[:values])	747✔
116	elsif request[:type] == :sync	760✔
117	self.synchronise_with_backend	771✔
118	end
119	rescue
120	$stderr.puts("ERROR: Edit mediator: #{$!}: #{$@}")	23✔
121	sleep 5	23✔
122	end
123	end
124	end
125
126
127	# A separate thread to trigger a sync with the backend every now and then.
128	Thread.new do	11✔
129	while true	11✔
130	sleep SYNC_WITH_BACKEND_SECONDS	782✔
131	@queue << {:type => :sync}	771✔
132	end
133	end
134	end
135	end
136
137
138	# When the system starts, run the update thread
139	UpdateThread.new(@active_edits, @queue).start	11✔
140	end

archivesspace / archivesspace / 19916213389

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous