Class: Takagi::Message::DeduplicationCache

Inherits:
Object
  • Object
show all
Defined in:
lib/takagi/message/deduplication_cache.rb,
sig/takagi/message/deduplication_cache.rbs

Overview

Implements message deduplication as per RFC 7252 Section 4.4

The server MUST detect duplicates by matching both Message ID and source endpoint. When a duplicate CON message is received, the server MUST resend the cached response.

Cache entries expire after EXCHANGE_LIFETIME (247 seconds per RFC 7252 §4.8.2)

Constant Summary collapse

EXCHANGE_LIFETIME =

RFC 7252 §4.8.2: EXCHANGE_LIFETIME = 247 seconds

Returns:

  • (247) 
247
CacheEntry =

Entry contains the cached response and metadata

Returns:

  • (Object) 

      
    

  







  
    

      Instance Method Summary
      collapse
    


    
    
  



  
Constructor Details

  
    

  

  
    #initialize  ⇒ DeduplicationCache 
  

  

  




  

    
Returns a new instance of DeduplicationCache.


  





  



  

    
      


22
23
24
25

    		
    
      
# File 'lib/takagi/message/deduplication_cache.rb', line 22

def initialize
  @cache = {}
  @mutex = Mutex.new
end

    		
  





  




  

    
Instance Method Details


    
      

  

  
    #cache_key(message_id, source_endpoint)  ⇒ ::String 
  

  

  




  

    

  





  
Parameters:


    
  
    
  • 
      
        message_id
      
      
        (Object)
      
      
      
    
    • 
  
    
  • 
      
        source_endpoint
      
      
        (Object)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (::String)
      
      
      
    
    • 
  





  

    
      


72
73
74

    		
    
      
# File 'lib/takagi/message/deduplication_cache.rb', line 72

def cache_key(message_id, source_endpoint)
  "#{source_endpoint}:#{message_id}"
end

    		
  





    
      

  

  
    #check_duplicate(message_id, source_endpoint)  ⇒ String? 
  

  

  




  

    
Check if message is a duplicate and return cached response if available


  





  
Parameters:


    
  
    
  • 
      
        message_id
      
      
        (Integer)
      
      
      
        —
        
    The CoAP Message ID
    
      
    
    • 
  
    
  • 
      
        source_endpoint
      
      
        (String)
      
      
      
        —
        
    Source IP:Port identifier
    
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (String, nil)
      
      
      
        —
        
    Cached response data or nil if not a duplicate
    
      
    
    • 
  





  

    
      


31
32
33
34
35
36
37
38

    		
    
      
# File 'lib/takagi/message/deduplication_cache.rb', line 31

def check_duplicate(message_id, source_endpoint)
  @mutex.synchronize do
    cleanup_expired_entries
    key = cache_key(message_id, source_endpoint)
    entry = @cache[key]
    entry&.response_data
  end
end

    		
  





    
      

  

  
    #cleanup_expired_entries  ⇒ Object 
  

  

  




  

    
Remove expired entries to prevent unbounded memory growth
RFC 7252 §4.8.2: entries older than EXCHANGE_LIFETIME should be discarded


  





  

Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


78
79
80
81

    		
    
      
# File 'lib/takagi/message/deduplication_cache.rb', line 78

def cleanup_expired_entries
  current_time = Time.now.to_f
  @cache.delete_if { |_key, entry| entry.expired?(current_time) }
end

    		
  





    
      

  

  
    #clear  ⇒ Object 
  

  

  




  

    
Clear all cache entries (useful for testing)


  





  

Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


56
57
58

    		
    
      
# File 'lib/takagi/message/deduplication_cache.rb', line 56

def clear
  @mutex.synchronize { @cache.clear }
end

    		
  





    
      

  

  
    #stats  ⇒ Object 
  

  

  




  

    
Get cache statistics


  





  

Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


61
62
63
64
65
66
67
68

    		
    
      
# File 'lib/takagi/message/deduplication_cache.rb', line 61

def stats
  @mutex.synchronize do
    {
      size: @cache.size,
      entries: @cache.keys
    }
  end
end

    		
  





    
      

  

  
    #store_response(message_id, source_endpoint, response_data)  ⇒ Object 
  

  

  




  

    
Store a response for future duplicate detection


  





  
Parameters:


    
  
    
  • 
      
        message_id
      
      
        (Integer)
      
      
      
        —
        
    The CoAP Message ID
    
      
    
    • 
  
    
  • 
      
        source_endpoint
      
      
        (String)
      
      
      
        —
        
    Source IP:Port identifier
    
      
    
    • 
  
    
  • 
      
        response_data
      
      
        (String)
      
      
      
        —
        
    The serialized response to cache
    
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (Object)
      
      
      
    
    • 
  





  

    
      


44
45
46
47
48
49
50
51
52
53

    		
    
      
# File 'lib/takagi/message/deduplication_cache.rb', line 44

def store_response(message_id, source_endpoint, response_data)
  @mutex.synchronize do
    key = cache_key(message_id, source_endpoint)
    @cache[key] = CacheEntry.new(
      response_data,
      Time.now.to_f,
      source_endpoint
    )
  end
end