<s> import logging <EOL> import threading <EOL> from hazelcast . hash import hash_to_index <EOL> from hazelcast . protocol . codec import client_get_partitions_codec <EOL> PARTITION_UPDATE_INTERVAL = <NUM_LIT:10> <EOL> class PartitionService ( object ) : <EOL> logger = logging . getLogger ( "<STR_LIT>" ) <EOL> timer = None <EOL> def __init__ ( self , client ) : <EOL> self . partitions = { } <EOL> self . _client = client <EOL> def start ( self ) : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> def partition_updater ( ) : <EOL> self . _do_refresh ( ) <EOL> self . timer = self . _client . reactor . add_timer ( PARTITION_UPDATE_INTERVAL , partition_updater ) <EOL> self . timer = self . _client . reactor . add_timer ( PARTITION_UPDATE_INTERVAL , partition_updater ) <EOL> def shutdown ( self ) : <EOL> if self . timer : <EOL> self . timer . cancel ( ) <EOL> def refresh ( self ) : <EOL> self . _client . reactor . add_timer ( <NUM_LIT:0> , self . _do_refresh ) <EOL> def get_partition_owner ( self , partition_id ) : <EOL> if partition_id not in self . partitions : <EOL> self . _do_refresh ( ) <EOL> return self . partitions [ partition_id ] <EOL> def get_partition_id ( self , key ) : <EOL> data = self . _client . serialization_service . to_data ( key ) <EOL> count = self . get_partition_count ( ) <EOL> return hash_to_index ( data . get_partition_hash ( ) , count ) <EOL> def get_partition_count ( self ) : <EOL> if not self . partitions : <EOL> self . _get_partition_count_blocking ( ) <EOL> return len ( self . partitions ) <EOL> def _get_partition_count_blocking ( self ) : <EOL> event = threading . Event ( ) <EOL> while not self . partitions : <EOL> self . _do_refresh ( callback = lambda : event . set ( ) ) <EOL> event . wait ( timeout = <NUM_LIT:1> ) <EOL> def _do_refresh ( self , callback = None ) : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> address = self . _client . cluster . owner_connection_address <EOL> connection = self . _client . connection_manager . get_connection ( address ) <EOL> if connection is None : <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> return <EOL> request = client_get_partitions_codec . encode_request ( ) <EOL> def cb ( f ) : <EOL> if f . is_success ( ) : <EOL> self . process_partition_response ( f . result ( ) ) <EOL> if callback : <EOL> callback ( ) <EOL> future = self . _client . invoker . invoke_on_connection ( request , connection ) <EOL> future . add_done_callback ( cb ) <EOL> def process_partition_response ( self , message ) : <EOL> partitions = client_get_partitions_codec . decode_response ( message ) [ "<STR_LIT>" ] <EOL> self . partitions = { } <EOL> for addr , partition_list in partitions . iteritems ( ) : <EOL> for partition in partition_list : <EOL> self . partitions [ partition ] = addr <EOL> self . logger . debug ( "<STR_LIT>" ) <EOL> def string_partition_strategy ( key ) : <EOL> if key is None : <EOL> return None <EOL> try : <EOL> index_of = key . index ( '<STR_LIT:@>' ) <EOL> return key [ index_of + <NUM_LIT:1> : ] <EOL> except ValueError : <EOL> return key </s>
