@Test public void testRelease1AtATime() throws Exception { final Timing timing = new Timing(); final int CLIENT_QTY = 10; final int MAX = CLIENT_QTY / 2; final AtomicInteger maxLeases = new AtomicInteger(0); final AtomicInteger activeQty = new AtomicInteger(0); final AtomicInteger uses = new AtomicInteger(0); List<Future<Object>> futures = Lists.newArrayList(); ExecutorService service = Executors.newFixedThreadPool(CLIENT_QTY); for ( int i = 0; i < CLIENT_QTY; ++i ) { Future<Object> f = service.submit ( new Callable<Object>() { @Override public Object call() throws Exception { CuratorFramework client = CuratorFrameworkFactory.newClient(server.getConnectString(), timing.session(), timing.connection(), new RetryOneTime(1)); client.start(); try { InterProcessSemaphoreV2 semaphore = new InterProcessSemaphoreV2(client, "/test", MAX); Lease lease = semaphore.acquire(timing.forWaiting().seconds(), TimeUnit.SECONDS); Assert.assertNotNull(lease); uses.incrementAndGet(); try { synchronized(maxLeases) { int qty = activeQty.incrementAndGet(); if ( qty > maxLeases.get() ) { maxLeases.set(qty); } } timing.sleepABit(); } finally { activeQty.decrementAndGet(); lease.close(); } } finally { TestCleanState.closeAndTestClean(client); } return null; } } ); futures.add(f); } for ( Future<Object> f : futures ) { f.get(); } Assert.assertEquals(uses.get(), CLIENT_QTY); Assert.assertEquals(maxLeases.get(), MAX); }
public Lease acquire() throws Exception { String path = internals.attemptLock(-1, null, null); return makeLease(path); }
InterProcessSemaphore { public Lease acquire() throws Exception { String path = internals.attemptLock(-1, null, null); return makeLease(path); } }
InterProcessSemaphore { public Lease acquire() throws Exception { String path = internals.attemptLock(-1, null, null); return makeLease(path); } InterProcessSemaphore(CuratorFramework client, String path, int maxLeases);  InterProcessSemaphore(CuratorFramework client, String path, SharedCountReader count); private  InterProcessSemaphore(CuratorFramework client, String path, int maxLeases, SharedCountReader count); }
InterProcessSemaphore { public Lease acquire() throws Exception { String path = internals.attemptLock(-1, null, null); return makeLease(path); } InterProcessSemaphore(CuratorFramework client, String path, int maxLeases);  InterProcessSemaphore(CuratorFramework client, String path, SharedCountReader count); private  InterProcessSemaphore(CuratorFramework client, String path, int maxLeases, SharedCountReader count); void returnAll(Collection<Lease> leases); void returnLease(Lease lease); Lease acquire(); Collection<Lease> acquire(int qty); Lease acquire(long time, TimeUnit unit); Collection<Lease> acquire(int qty, long time, TimeUnit unit); }
InterProcessSemaphore { public Lease acquire() throws Exception { String path = internals.attemptLock(-1, null, null); return makeLease(path); } InterProcessSemaphore(CuratorFramework client, String path, int maxLeases);  InterProcessSemaphore(CuratorFramework client, String path, SharedCountReader count); private  InterProcessSemaphore(CuratorFramework client, String path, int maxLeases, SharedCountReader count); void returnAll(Collection<Lease> leases); void returnLease(Lease lease); Lease acquire(); Collection<Lease> acquire(int qty); Lease acquire(long time, TimeUnit unit); Collection<Lease> acquire(int qty, long time, TimeUnit unit);  }