Sha256: 8a8af16d86d2d8b79d41b8d8bfa01293dfdc312a047345f8ab8b663f1072d1b4
Contents?: true
Size: 1.43 KB
Versions: 8
Compression:
Stored size: 1.43 KB
Contents
# frozen_string_literal: true
require_relative 'periodic_sync'
module Prefab
class ContextShapeAggregator
include Prefab::PeriodicSync
LOG = Prefab::InternalLogger.new(self)
attr_reader :data
def initialize(client:, max_shapes:, sync_interval:)
@max_shapes = max_shapes
@client = client
@name = 'context_shape_aggregator'
@data = Concurrent::Set.new
start_periodic_sync(sync_interval)
end
def push(context)
return if @data.size >= @max_shapes
context.contexts.each_pair do |name, name_context|
name_context.to_h.each_pair do |key, value|
@data.add [name, key, Prefab::ContextShape.field_type_number(value)]
end
end
end
def prepare_data
duped = @data.dup
@data.clear
duped.inject({}) do |acc, (name, key, type)|
acc[name] ||= {}
acc[name][key] = type
acc
end
end
private
def flush(to_ship, _)
pool.post do
LOG.debug "Uploading context shapes for #{to_ship.values.size}"
shapes = PrefabProto::ContextShapes.new(
shapes: to_ship.map do |name, shape|
PrefabProto::ContextShape.new(
name: name,
field_types: shape
)
end
)
result = post('/api/v1/context-shapes', shapes)
LOG.debug "Uploaded #{to_ship.values.size} shapes: #{result.status}"
end
end
end
end
Version data entries
8 entries across 8 versions & 1 rubygems