Class: SolidMCP::MessageWriter

Inherits:
Object
  • Object
show all
Includes:
Singleton
Defined in:
lib/solid_mcp/message_writer.rb

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ MessageWriter

Returns a new instance of MessageWriter.



15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
 
# File 'lib/solid_mcp/message_writer.rb', line 15

def initialize
  @queue = SizedQueue.new(SolidMCP.configuration.max_queue_size)
  @shutdown = Concurrent::AtomicBoolean.new(false)
  @dropped_count = Concurrent::AtomicFixnum.new(0)
  @worker_ready = Concurrent::CountDownLatch.new(1)
  @executor = Concurrent::ThreadPoolExecutor.new(
    min_threads: 1,
    max_threads: 1,  # Single thread for ordered writes
    max_queue: 0,    # Unbounded queue
    fallback_policy: :caller_runs
  )
  start_worker
  # Wait for worker thread to be ready (with short timeout)
  # Using 0.1s is enough for worker to start, avoids 1s delay per test
  @worker_ready.wait(0.1)
end

Class Method Details

.reset! ⇒ Object

Reset the singleton (for testing only)



11
12
13
 
# File 'lib/solid_mcp/message_writer.rb', line 11

def self.reset!
  @singleton__instance__ = nil
end

Instance Method Details

#dropped_count ⇒ Object

Get count of dropped messages



54
55
56
 
# File 'lib/solid_mcp/message_writer.rb', line 54

def dropped_count
  @dropped_count.value
end

#enqueue(session_id, event_type, data) ⇒ Object

Called by publish API - non-blocking with backpressure



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/solid_mcp/message_writer.rb', line 33

def enqueue(session_id, event_type, data)
  message = {
    session_id: session_id,
    event_type: event_type,
    data: data.is_a?(String) ? data : data.to_json,
    created_at: Time.now.utc
  }

  # Try non-blocking push with backpressure
  begin
    @queue.push(message, true) # non-blocking
    true
  rescue ThreadError
    # Queue full - drop message and log
    @dropped_count.increment
    SolidMCP::Logger.warn "SolidMCP queue full (#{SolidMCP.configuration.max_queue_size}), dropped message for session #{session_id}"
    false
  end
end

#flush ⇒ Object

Force flush any pending messages (useful for tests)



75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/solid_mcp/message_writer.rb', line 75

def flush
  return unless @executor.running?

  # Add a marker and wait for it to be processed
  processed = Concurrent::CountDownLatch.new(1)

  # Use blocking push for flush marker (not subject to queue limits)
  begin
    @queue.push({ flush_marker: processed }, false) # blocking
  rescue ThreadError
    # Queue is shutting down
    return
  end

  # Wait up to 1 second for flush to complete
  processed.wait(1)
end

#shutdown ⇒ Object

Blocks until executor has flushed everything



59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/solid_mcp/message_writer.rb', line 59

def shutdown
  SolidMCP::Logger.info "SolidMCP::MessageWriter shutting down, #{@queue.size} messages pending"

  # Mark as shutting down (worker will exit after draining queue)
  @shutdown.make_true

  # Wait for executor to finish processing
  @executor.shutdown
  @executor.wait_for_termination(SolidMCP.configuration.shutdown_timeout)

  if @queue.size > 0
    SolidMCP::Logger.warn "SolidMCP::MessageWriter shutdown timeout, #{@queue.size} messages not written"
  end
end