Show More
@@ -23,7 +23,7 class AsyncResult(object): | |||
|
23 | 23 | """ |
|
24 | 24 | def __init__(self, client, msg_ids): |
|
25 | 25 | self._client = client |
|
26 |
self. |
|
|
26 | self.msg_ids = msg_ids | |
|
27 | 27 | self._ready = False |
|
28 | 28 | self._success = None |
|
29 | 29 | |
@@ -31,7 +31,7 class AsyncResult(object): | |||
|
31 | 31 | if self._ready: |
|
32 | 32 | return "<%s: finished>"%(self.__class__.__name__) |
|
33 | 33 | else: |
|
34 |
return "<%s: %r>"%(self.__class__.__name__,self. |
|
|
34 | return "<%s: %r>"%(self.__class__.__name__,self.msg_ids) | |
|
35 | 35 | |
|
36 | 36 | |
|
37 | 37 | def _reconstruct_result(self, res): |
@@ -74,10 +74,10 class AsyncResult(object): | |||
|
74 | 74 | """ |
|
75 | 75 | if self._ready: |
|
76 | 76 | return |
|
77 |
self._ready = self._client.barrier(self. |
|
|
77 | self._ready = self._client.barrier(self.msg_ids, timeout) | |
|
78 | 78 | if self._ready: |
|
79 | 79 | try: |
|
80 |
results = map(self._client.results.get, self. |
|
|
80 | results = map(self._client.results.get, self.msg_ids) | |
|
81 | 81 | results = error.collect_exceptions(results, 'get') |
|
82 | 82 | self._result = self._reconstruct_result(results) |
|
83 | 83 | except Exception, e: |
@@ -504,7 +504,7 class Client(object): | |||
|
504 | 504 | if isinstance(msg_id, int): |
|
505 | 505 | msg_id = self.history[msg_id] |
|
506 | 506 | elif isinstance(msg_id, AsyncResult): |
|
507 |
map(theids.add, msg_id. |
|
|
507 | map(theids.add, msg_id.msg_ids) | |
|
508 | 508 | continue |
|
509 | 509 | theids.add(msg_id) |
|
510 | 510 | if not theids.intersection(self.outstanding): |
@@ -849,7 +849,7 class Client(object): | |||
|
849 | 849 | r = self.push({key: partition[0]}, targets=engineid, block=False) |
|
850 | 850 | else: |
|
851 | 851 | r = self.push({key: partition}, targets=engineid, block=False) |
|
852 |
msg_ids.extend(r. |
|
|
852 | msg_ids.extend(r.msg_ids) | |
|
853 | 853 | r = AsyncResult(self, msg_ids) |
|
854 | 854 | if block: |
|
855 | 855 | return r.get() |
@@ -866,7 +866,7 class Client(object): | |||
|
866 | 866 | mapObject = Map.dists[dist]() |
|
867 | 867 | msg_ids = [] |
|
868 | 868 | for index, engineid in enumerate(targets): |
|
869 |
msg_ids.extend(self.pull(key, targets=engineid,block=False). |
|
|
869 | msg_ids.extend(self.pull(key, targets=engineid,block=False).msg_ids) | |
|
870 | 870 | |
|
871 | 871 | r = AsyncMapResult(self, msg_ids, mapObject) |
|
872 | 872 | if block: |
@@ -344,4 +344,12 class LoadBalancedView(View): | |||
|
344 | 344 | def __repr__(self): |
|
345 | 345 | return "<%s %s>"%(self.__class__.__name__, self.client._addr) |
|
346 | 346 | |
|
347 | @property | |
|
348 | def targets(self): | |
|
349 | return None | |
|
350 | ||
|
351 | @targets.setter | |
|
352 | def targets(self, value): | |
|
353 | raise AttributeError("Cannot set targets for LoadbalancedView!") | |
|
354 | ||
|
347 | 355 | No newline at end of file |
General Comments 0
You need to be logged in to leave comments.
Login now