Class: AceMQ::AMQP::Patterns::FilesystemClaimCheckStore

Inherits:
Object
  • Object
show all
Defined in:
lib/acemq/amqp/patterns/claim_check.rb

Overview

A claim-check store on a filesystem.

Useful where the filesystem is shared and durable — an NFS mount, a persistent volume — and the honest middle ground between a hash and object storage. On a container's local disk it is InMemoryClaimCheckStore with extra steps: the consumer is on another host and finds nothing.

Object storage is the usual right answer, and a store in front of S3 or Azure Blob Storage is three short methods. This one exists because "write it to the mount everything already has" is a real deployment and not a bad one.

Writes are atomic

The payload is written to a temporary file and renamed into place. Without that, a consumer fast enough to read the key before the writer finished gets a truncated payload and a parse error somewhere unhelpful — and messaging is exactly the arrangement that makes a consumer that fast normal rather than unlikely. rename within one directory is atomic, so a reader sees the whole payload or no payload.

A filesystem that cannot be written to raises whatever the filesystem raised, unwrapped: Errno::EACCES already names the path and the problem, and a library exception in front of it would only hide both.

Constant Summary collapse

SAFE_KEY =

A key reaches the filesystem as a path segment, so it is checked rather than trusted. Every key this store issues is a UUID; one arriving from a message is whatever a publisher put there, and ../../etc/passwd is a key too.

/\A[A-Za-z0-9][A-Za-z0-9._-]{0,127}\z/

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(directory) ⇒ FilesystemClaimCheckStore

Returns a new instance of FilesystemClaimCheckStore.

Parameters:

  • directory (String) —

    where payloads are written; created if it is not there



164
165
166
167
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 164

def initialize(directory)
  @directory = directory.to_s
  FileUtils.mkdir_p(@directory)
end

Instance Attribute Details

#directory ⇒ Object (readonly)

The directory payloads are written to.



170
171
172
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 170

def directory
  @directory
end

Instance Method Details

#delete(key) ⇒ Object

Removes a payload.



209
210
211
212
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 209

def delete(key)
  FileUtils.rm_f(path_for(key))
  nil
end

#get(key) ⇒ String?

Redeems a claim check.

Parameters:

  • key (String) —

    what the message carried

Returns:

  • (String, nil) —

    the payload, or nil when the store no longer holds it



199
200
201
202
203
204
205
206
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 199

def get(key)
  File.binread(path_for(key))
rescue Errno::ENOENT
  # Absent rather than failed: a key the store no longer holds is a
  # retention answer, and the codec turns it into a message that
  # explains itself.
  nil
end

#put(content) ⇒ String

Stores a payload, atomically.

Parameters:

  • content (String) —

    the bytes

Returns:

  • (String) —

    the key the message will carry



176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/acemq/amqp/patterns/claim_check.rb', line 176

def put(content)
  key = SecureRandom.uuid
  staging = File.join(@directory, "#{key}.partial")
  begin
    File.binwrite(staging, content.to_s)
    # Moved into place rather than written in place, so a reader either
    # finds the whole payload or finds nothing.
    File.rename(staging, File.join(@directory, key))
  ensure
    # A failed write leaves a .partial nobody will ever claim. Removing
    # it here is the difference between a directory that fills with
    # debris and one that does not, and rm_f on a name the rename has
    # already taken away is a no-op rather than an error.
    FileUtils.rm_f(staging)
  end
  key
end