diff options
Diffstat (limited to 'java-compat/src')
| -rw-r--r-- | java-compat/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java | 197 | ||||
| -rw-r--r-- | java-compat/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java | 129 |
2 files changed, 241 insertions, 85 deletions
diff --git a/java-compat/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java b/java-compat/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java index ca6407aff70..e5a35ee7bfd 100644 --- a/java-compat/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java +++ b/java-compat/src/IceDiscovery/src/main/java/IceDiscovery/LookupI.java @@ -21,7 +21,7 @@ class LookupI extends _LookupDisp Request(T id, int retryCount) { _id = id; - _nRetry = retryCount; + _retryCount = retryCount; } T @@ -40,7 +40,29 @@ class LookupI extends _LookupDisp boolean retry() { - return --_nRetry >= 0; + return --_retryCount >= 0; + } + + void + invoke(String domainId, Map<LookupPrx, LookupReplyPrx> lookups) + { + _lookupCount = lookups.size(); + _failureCount = 0; + for(Map.Entry<LookupPrx, LookupReplyPrx> entry : lookups.entrySet()) + { + invokeWithLookup(domainId, entry.getKey(), entry.getValue()); + } + } + + boolean + exception() + { + if(++_failureCount == _lookupCount) + { + finished(null); + return true; + } + return false; } void @@ -57,9 +79,15 @@ class LookupI extends _LookupDisp _future = null; } - protected int _nRetry; + abstract void finished(Ice.ObjectPrx proxy); + + abstract protected void invokeWithLookup(String domainId, LookupPrx lookup, LookupReplyPrx lookupReply); + + protected int _retryCount; + protected int _lookupCount; + protected int _failureCount; protected List<AmdCB> _callbacks = new ArrayList<AmdCB>(); - private T _id; + protected T _id; protected java.util.concurrent.Future<?> _future; }; @@ -76,7 +104,7 @@ class LookupI extends _LookupDisp boolean retry() { - return _proxies.size() == 0 && --_nRetry >= 0; + return _proxies.size() == 0 && --_retryCount >= 0; } boolean @@ -101,6 +129,7 @@ class LookupI extends _LookupDisp return true; } + @Override void finished(Ice.ObjectPrx proxy) { @@ -135,6 +164,28 @@ class LookupI extends _LookupDisp adapterRequestTimedOut(this); } + @Override + protected void + invokeWithLookup(String domainId, LookupPrx lookup, LookupReplyPrx lookupReply) + { + lookup.begin_findAdapterById(domainId, _id, lookupReply, new Ice.Callback() + { + @Override + public void + completed(Ice.AsyncResult r) + { + try + { + r.throwLocalException(); + } + catch(Ice.LocalException ex) + { + adapterRequestException(AdapterRequest.this, ex); + } + } + }); + } + private void sendResponse(Ice.ObjectPrx proxy) { @@ -163,6 +214,7 @@ class LookupI extends _LookupDisp finished(proxy); } + @Override void finished(Ice.ObjectPrx proxy) { @@ -179,32 +231,40 @@ class LookupI extends _LookupDisp { objectRequestTimedOut(this); } + + @Override + protected void + invokeWithLookup(String domainId, LookupPrx lookup, LookupReplyPrx lookupReply) + { + lookup.begin_findObjectById(domainId, _id, lookupReply, new Ice.Callback() + { + @Override + public void + completed(Ice.AsyncResult r) + { + try + { + r.throwLocalException(); + } + catch(Ice.LocalException ex) + { + objectRequestException(ObjectRequest.this, ex); + } + } + }); + } }; public LookupI(LocatorRegistryI registry, LookupPrx lookup, Ice.Properties properties) { _registry = registry; + _lookup = lookup; _timeout = properties.getPropertyAsIntWithDefault("IceDiscovery.Timeout", 300); _retryCount = properties.getPropertyAsIntWithDefault("IceDiscovery.RetryCount", 3); _latencyMultiplier = properties.getPropertyAsIntWithDefault("IceDiscovery.LatencyMultiplier", 1); _domainId = properties.getProperty("IceDiscovery.DomainId"); _timer = IceInternal.Util.getInstance(lookup.ice_getCommunicator()).timer(); - try - { - lookup.ice_getConnection(); - } - catch(Ice.LocalException ex) - { - StringBuilder b = new StringBuilder(); - b.append("IceDiscovery is unable to establish a multicast connection:\n"); - b.append("proxy = "); - b.append(lookup.toString()); - b.append('\n'); - b.append(ex.toString()); - throw new Ice.PluginInitializationException(b.toString()); - } - // // Create one lookup proxy per endpoint from the given proxy. We want to send a multicast // datagram on each endpoint. @@ -212,18 +272,10 @@ class LookupI extends _LookupDisp Ice.Endpoint[] single = new Ice.Endpoint[1]; for(Ice.Endpoint endpt : lookup.ice_getEndpoints()) { - try - { - single[0] = endpt; - LookupPrx l = (LookupPrx)lookup.ice_endpoints(single); - l.ice_getConnection(); - _lookup.put(l, null); - } - catch(Ice.LocalException ex) - { - } + single[0] = endpt; + _lookups.put((LookupPrx)lookup.ice_endpoints(single), null); } - assert(!_lookup.isEmpty()); + assert(!_lookups.isEmpty()); } void @@ -233,7 +285,7 @@ class LookupI extends _LookupDisp // Use a lookup reply proxy whose adress matches the interface used to send multicast datagrams. // Ice.Endpoint[] single = new Ice.Endpoint[1]; - for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookup.entrySet()) + for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookups.entrySet()) { Ice.UDPEndpointInfo info = (Ice.UDPEndpointInfo)entry.getKey().ice_getEndpoints()[0].getInfo(); if(!info.mcastInterface.isEmpty()) @@ -324,10 +376,7 @@ class LookupI extends _LookupDisp { try { - for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookup.entrySet()) - { - entry.getKey().begin_findObjectById(_domainId, id, entry.getValue()); - } + request.invoke(_domainId, _lookups); request.scheduleTimer(_timeout); } catch(Ice.LocalException ex) @@ -352,10 +401,7 @@ class LookupI extends _LookupDisp { try { - for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookup.entrySet()) - { - entry.getKey().begin_findAdapterById(_domainId, adapterId, entry.getValue()); - } + request.invoke(_domainId, _lookups); request.scheduleTimer(_timeout); } catch(Ice.LocalException ex) @@ -409,10 +455,7 @@ class LookupI extends _LookupDisp { try { - for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookup.entrySet()) - { - entry.getKey().begin_findObjectById(_domainId, request.getId(), entry.getValue()); - } + request.invoke(_domainId, _lookups); request.scheduleTimer(_timeout); return; } @@ -426,6 +469,34 @@ class LookupI extends _LookupDisp } synchronized void + objectRequestException(ObjectRequest request, Ice.LocalException ex) + { + ObjectRequest r = _objectRequests.get(request.getId()); + if(r == null || r != request) + { + return; + } + + if(request.exception()) + { + if(_warnOnce) + { + StringBuilder s = new StringBuilder(); + s.append("failed to lookup object `"); + s.append(_lookup.ice_getCommunicator().identityToString(request.getId())); + s.append("' with lookup proxy `"); + s.append(_lookup); + s.append("':\n"); + s.append(ex.toString()); + _lookup.ice_getCommunicator().getLogger().warning(s.toString()); + _warnOnce = false; + } + request.cancelTimer(); + _objectRequests.remove(request.getId()); + } + } + + synchronized void adapterRequestTimedOut(AdapterRequest request) { AdapterRequest r = _adapterRequests.get(request.getId()); @@ -438,10 +509,7 @@ class LookupI extends _LookupDisp { try { - for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookup.entrySet()) - { - entry.getKey().begin_findAdapterById(_domainId, request.getId(), entry.getValue()); - } + request.invoke(_domainId, _lookups); request.scheduleTimer(_timeout); return; } @@ -454,14 +522,43 @@ class LookupI extends _LookupDisp _adapterRequests.remove(request.getId()); } + synchronized void + adapterRequestException(AdapterRequest request, Ice.LocalException ex) + { + AdapterRequest r = _adapterRequests.get(request.getId()); + if(r == null || r != request) + { + return; + } + + if(request.exception()) + { + if(_warnOnce) + { + StringBuilder s = new StringBuilder(); + s.append("failed to lookup adapter `"); + s.append(request.getId()); + s.append("' with lookup proxy `"); + s.append(_lookup); + s.append("':\n"); + s.append(ex.toString()); + _lookup.ice_getCommunicator().getLogger().warning(s.toString()); + _warnOnce = false; + } + request.cancelTimer(); + _adapterRequests.remove(request.getId()); + } + } + private LocatorRegistryI _registry; - private java.util.Map<LookupPrx, LookupReplyPrx> _lookup = new java.util.HashMap<>(); + private LookupPrx _lookup; + private java.util.Map<LookupPrx, LookupReplyPrx> _lookups = new java.util.HashMap<>(); private final int _timeout; private final int _retryCount; private final int _latencyMultiplier; private final String _domainId; - private final java.util.concurrent.ScheduledExecutorService _timer; + private boolean _warnOnce = true; private Map<Ice.Identity, ObjectRequest> _objectRequests = new HashMap<Ice.Identity, ObjectRequest>(); private Map<String, AdapterRequest> _adapterRequests = new HashMap<String, AdapterRequest>(); diff --git a/java-compat/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java b/java-compat/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java index 2b49d63747d..571a840c975 100644 --- a/java-compat/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java +++ b/java-compat/src/IceLocatorDiscovery/src/main/java/IceLocatorDiscovery/PluginI.java @@ -145,6 +145,7 @@ class PluginI implements Plugin LocatorI(String name, LookupPrx lookup, Ice.Properties properties, String instanceName, Ice.LocatorPrx voidLocator) { + _lookup = lookup; _timeout = properties.getPropertyAsIntWithDefault(name + ".Timeout", 300); _retryCount = properties.getPropertyAsIntWithDefault(name + ".RetryCount", 3); _retryDelay = properties.getPropertyAsIntWithDefault(name + ".RetryDelay", 2000); @@ -154,21 +155,8 @@ class PluginI implements Plugin _locator = lookup.ice_getCommunicator().getDefaultLocator(); _voidLocator = voidLocator; _pendingRetryCount = 0; - - try - { - lookup.ice_getConnection(); - } - catch(Ice.LocalException ex) - { - StringBuilder b = new StringBuilder(); - b.append("IceDiscovery is unable to establish a multicast connection:\n"); - b.append("proxy = "); - b.append(lookup.toString()); - b.append('\n'); - b.append(ex.toString()); - throw new Ice.PluginInitializationException(b.toString()); - } + _failureCount = 0; + _warnOnce = true; // // Create one lookup proxy per endpoint from the given proxy. We want to send a multicast @@ -177,18 +165,10 @@ class PluginI implements Plugin Ice.Endpoint[] single = new Ice.Endpoint[1]; for(Ice.Endpoint endpt : lookup.ice_getEndpoints()) { - try - { - single[0] = endpt; - LookupPrx l = (LookupPrx)lookup.ice_endpoints(single); - l.ice_getConnection(); - _lookup.put(l, null); - } - catch(Ice.LocalException ex) - { - } + single[0] = endpt; + _lookups.put((LookupPrx)lookup.ice_endpoints(single), null); } - assert(!_lookup.isEmpty()); + assert(!_lookups.isEmpty()); } public void @@ -198,7 +178,7 @@ class PluginI implements Plugin // Use a lookup reply proxy whose adress matches the interface used to send multicast datagrams. // Ice.Endpoint[] single = new Ice.Endpoint[1]; - for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookup.entrySet()) + for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookups.entrySet()) { Ice.UDPEndpointInfo info = (Ice.UDPEndpointInfo)entry.getKey().ice_getEndpoints()[0].getInfo(); if(!info.mcastInterface.isEmpty()) @@ -401,11 +381,26 @@ class PluginI implements Plugin if(_pendingRetryCount == 0) // No request in progress { _pendingRetryCount = _retryCount; + _failureCount = 0; try { - for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookup.entrySet()) + for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookups.entrySet()) { - entry.getKey().begin_findLocator(_instanceName, entry.getValue()); // Send multicast request + entry.getKey().begin_findLocator(_instanceName, entry.getValue(), new Ice.Callback() { + @Override + public void + completed(Ice.AsyncResult r) + { + try + { + r.throwLocalException(); + } + catch(Ice.LocalException ex) + { + exception(ex); + } + } + }); // Send multicast request } _future = _timer.schedule(_retryTask, _timeout, java.util.concurrent.TimeUnit.MILLISECONDS); } @@ -422,6 +417,45 @@ class PluginI implements Plugin } } + synchronized void + exception(Ice.LocalException ex) + { + if(++_failureCount == _lookups.size() && _pendingRetryCount > 0) + { + // + // All the lookup calls failed, cancel the timer and propagate the error to the requests. + // + _future.cancel(false); + _future = null; + + _pendingRetryCount = 0; + + if(_warnOnce) + { + StringBuilder builder = new StringBuilder(); + builder.append("failed to lookup locator with lookup proxy `"); + builder.append(_lookup); + builder.append("':\n"); + builder.append(ex); + _lookup.ice_getCommunicator().getLogger().warning(builder.toString()); + _warnOnce = false; + } + + if(_pendingRequests.isEmpty()) + { + notify(); + } + else + { + for(Request req : _pendingRequests) + { + req.invoke(_voidLocator); + } + _pendingRequests.clear(); + } + } + } + private Runnable _retryTask = new Runnable() { @Override @@ -433,9 +467,24 @@ class PluginI implements Plugin { try { - for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookup.entrySet()) + _failureCount = 0; + for(Map.Entry<LookupPrx, LookupReplyPrx> entry : _lookups.entrySet()) { - entry.getKey().begin_findLocator(_instanceName, entry.getValue()); // Send multicast request + entry.getKey().begin_findLocator(_instanceName, entry.getValue(), new Ice.Callback() { + @Override + public void + completed(Ice.AsyncResult r) + { + try + { + r.throwLocalException(); + } + catch(Ice.LocalException ex) + { + exception(ex); + } + } + }); // Send multicast request } _future = _timer.schedule(_retryTask, _timeout, java.util.concurrent.TimeUnit.MILLISECONDS); return; @@ -446,18 +495,26 @@ class PluginI implements Plugin _pendingRetryCount = 0; } - for(Request req : _pendingRequests) + if(_pendingRequests.isEmpty()) { - req.invoke(_voidLocator); + notify(); + } + else + { + for(Request req : _pendingRequests) + { + req.invoke(_voidLocator); + } + _pendingRequests.clear(); } - _pendingRequests.clear(); _nextRetry = IceInternal.Time.currentMonotonicTimeMillis() + _retryDelay; } } }; - private final java.util.Map<LookupPrx, LookupReplyPrx> _lookup = new java.util.HashMap<>(); + private final LookupPrx _lookup; + private final java.util.Map<LookupPrx, LookupReplyPrx> _lookups = new java.util.HashMap<>(); private final int _timeout; private java.util.concurrent.Future<?> _future; private final java.util.concurrent.ScheduledExecutorService _timer; @@ -471,6 +528,8 @@ class PluginI implements Plugin private Map<String, Ice.LocatorPrx> _locators = new java.util.HashMap<>(); private int _pendingRetryCount; + private int _failureCount; + private boolean _warnOnce; private List<Request> _pendingRequests = new ArrayList<Request>(); private long _nextRetry; }; |