diff --git a/spec/ldclient_listeners_spec.rb b/spec/ldclient_listeners_spec.rb index 2542b81e..927ffba7 100644 --- a/spec/ldclient_listeners_spec.rb +++ b/spec/ldclient_listeners_spec.rb @@ -13,8 +13,8 @@ module LaunchDarkly end it "sends status updates" do - store = MockBigSegmentStore.new - store.setup_metadata(Time.now) + store = GatedBigSegmentStore.new + store.queue_metadata(Time.now) big_segments_config = BigSegmentsConfig.new( store: store, status_poll_interval: 0.01 @@ -24,17 +24,29 @@ module LaunchDarkly expect(status1.available).to be(true) expect(status1.stale).to be(false) + # Status reads after the first come from the cache, so the second query comes from the + # poller. Observers have been notified of the first status by the time it starts, so the + # observer added below sees only later changes. + expect(store.wait_for_queries(2)).to be(true) + statuses = Queue.new observer = SimpleObserver.adding_to_queue(statuses) client.big_segment_store_status_provider.add_observer(observer) - store.setup_metadata_error(StandardError.new("sorry")) + # Fail the waiting query, then wait for the poller to start the next one. It has notified + # observers of the failure by then. + store.queue_metadata_error(StandardError.new("sorry")) + expect(store.wait_for_queries(3)).to be(true) + # The observer should have received one status, reporting the store as unavailable. + expect(statuses.size).to eq(1) status2 = statuses.pop expect(status2.available).to be(false) expect(status2.stale).to be(false) expect(client.big_segment_store_status_provider.status).to eq(status2) + ensure + store.release end end end diff --git a/spec/mock_components.rb b/spec/mock_components.rb index c7dc8b9b..c777de10 100644 --- a/spec/mock_components.rb +++ b/spec/mock_components.rb @@ -116,6 +116,87 @@ def setup_segment_for_context(user_key, segment, included) end end + # + # A Big Segment store whose metadata queries wait for the test to supply each result. + # + # The status poller notifies observers of one result before it starts its next query. Once a + # test sees a query start, every earlier result has been delivered, and the poller stays in that + # query until the test supplies another result. + # + # Call {#release} before closing the client. The poller cannot stop while a query is waiting. + # + class GatedBigSegmentStore + def initialize + @lock = Mutex.new + @changed = ConditionVariable.new + @results = [] + @last_result = nil + @queries_started = 0 + @released = false + end + + def get_metadata + result = @lock.synchronize do + @queries_started += 1 + @changed.broadcast + @changed.wait(@lock) while @results.empty? && !@released + @last_result = @results.shift unless @results.empty? + @last_result + end + raise result if result.is_a?(Exception) + result + end + + def get_membership(_context_hash) + nil + end + + def stop + end + + # Supplies the result of the next metadata query. + def queue_metadata(last_up_to_date) + queue_result(Interfaces::BigSegmentStoreMetadata.new(last_up_to_date.to_f * 1000)) + end + + # Makes the next metadata query raise the given error. + def queue_metadata_error(ex) + queue_result(ex) + end + + # + # Blocks until the store has received `count` metadata queries, or until the timeout passes. + # + # @return [Boolean] false if the queries did not start within the timeout + # + def wait_for_queries(count, timeout: 5) + deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout + @lock.synchronize do + while @queries_started < count + remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC) + return false if remaining <= 0 + @changed.wait(@lock, remaining) + end + true + end + end + + # Stops queries from waiting. Waiting and later queries return the last result again. + def release + @lock.synchronize do + @released = true + @changed.broadcast + end + end + + private def queue_result(result) + @lock.synchronize do + @results << result + @changed.broadcast + end + end + end + class SimpleObserver def initialize(fn) @fn = fn