class Movie::ClusterShardingExtension

• Movie::ClusterShardingExtension
• Movie::Extension
• Reference
• Object

Defined in:

movie/cluster/sharding/coordination.cr
movie/cluster/sharding/extension.cr
movie/cluster/sharding/routing.cr
movie/persistence/sharding.cr

Constant Summary

CONTROL_ACK_TAG = "movie.cluster.sharding.control-ack.v1"

DAEMON_NAME = "sharding"

DELIVERY_RETRY_DRAIN_BATCH = 64

HANDOFF_BUFFER_CAPACITY = 1024

HANDOFF_TIMEOUT = 2.seconds

LEASE_RETRY_INTERVAL = 25.milliseconds

Log = ::Log.for(self)

PLAN_SYNC_INTERVAL = 250.milliseconds

PROTOCOL_TAG = "movie.cluster.sharding.envelope.v1"

RECONCILE_INTERVAL = 100.milliseconds

Constructors

• .new(system : AbstractActorSystem)

Instance Method Summary

• #activate(entity_type : Cluster::ShardedEntityType(T), entity_id : String, timeout : Time::Span = 2.seconds) : Future(Cluster::ShardingControlAck) forall T

  Eagerly creates one entity through the authoritative shard coordinator.

• #allocations(entity_type : Cluster::ShardedEntityType(T)) : Cluster::ShardAllocations forall T
• #entity_ref_for(entity_type : Cluster::ShardedEntityType(T), entity_id : String) : Cluster::ShardedEntityRef(T) forall T
• #handle_cluster_event(event : Cluster::ClusterEvent) : Nil
• #handle_envelope(envelope : Cluster::ShardingEnvelope, sender : ActorRefBase | Nil, remote_address : Address | Nil, remote_node_uid : String | Nil) : Nil
• #handoff_in_progress?(entity_type : Cluster::ShardedEntityType(T), entity_id : String) : Bool forall T
• #handoffs_in_progress : Int32
• #init(name : String, message_type : T.class, shard_count : Int32 = 256, partitioner : Cluster::EntityPartitioner = Cluster::StableHashPartitioner.new, allocation : Cluster::ShardAllocationStrategy = Cluster::LeastLoadedAllocation.new, rebalance : Cluster::RebalancePolicy = Cluster::RateLimitedRebalance.new, idle_timeout : Time::Span | Nil = nil, &factory : String -> AbstractBehavior(T)) : Cluster::ShardedEntityType(T) forall T
• #init_durable_state(entity_type : Persistence::EntityType(T), shard_count : Int32 = 256, partitioner : Cluster::EntityPartitioner = Cluster::StableHashPartitioner.new, allocation : Cluster::ShardAllocationStrategy = Cluster::LeastLoadedAllocation.new, rebalance : Cluster::RebalancePolicy = Cluster::RateLimitedRebalance.new, lease_duration : Time::Span = 10.seconds, lease_renew_interval : Time::Span = 3.seconds, idle_timeout : Time::Span | Nil = nil, routing_name : String = entity_type.name) : Cluster::ShardedEntityType(T) forall T

  routing_name has the same compatibility and isolation semantics as the event-sourced overload above.

• #init_event_sourced(entity_type : Persistence::EntityType(T), shard_count : Int32 = 256, partitioner : Cluster::EntityPartitioner = Cluster::StableHashPartitioner.new, allocation : Cluster::ShardAllocationStrategy = Cluster::LeastLoadedAllocation.new, rebalance : Cluster::RebalancePolicy = Cluster::RateLimitedRebalance.new, lease_duration : Time::Span = 10.seconds, lease_renew_interval : Time::Span = 3.seconds, idle_timeout : Time::Span | Nil = nil, routing_name : String = entity_type.name) : Cluster::ShardedEntityType(T) forall T

  routing_name is an advanced logical facade name.

• #local_entity_count : Int32
• #owner_for(entity_type : Cluster::ShardedEntityType(T), entity_id : String) : Cluster::UniqueAddress | Nil forall T

  Returns the current planned owner for a typed entity, or nil while its shard has no eligible owner or no allocation plan has been observed.

• #passivate(entity_type : String, entity_id : String) : Bool
• #route(entity_type : String, entity_id : String, message : T, sender : ActorRefBase | Nil = nil) : Nil forall T
• #route_ask(entity_type : String, entity_id : String, message : T, response_type : R.class, timeout : Time::Span) : Future(R) forall T, R
• #start : Bool

  Called when the extension is registered with the system.

• #stats : Cluster::ShardingStats
• #stop : Nil

  Called when the system is shutting down.

Constructor Detail

Instance Method Detail