import com.twitter.util.Try; //导入依赖的package包/类
@Override public void sendSpans(final List<byte[]> spans, final Callback callback) {
  Trace.letClear(new AbstractFunction0<Void>() {
    @Override public Void apply() {
      try {
        if (closeCalled) throw new IllegalStateException("closed");
        client.apply(makeRequest(spans)).respond(new AbstractFunction1<Try<Rep>, BoxedUnit>() {
          @Override public BoxedUnit apply(Try<Rep> result) {
            if (result.isReturn()) {
              callback.onComplete();
            } else {
              callback.onError(result.throwable());
            }
            return BoxedUnit.UNIT;
          }
        });
      } catch (Throwable e) {
        callback.onError(e);
        if (e instanceof Error) throw (Error) e;
      }
      return null;
    }
  });
}
 

import com.twitter.util.Try; //导入依赖的package包/类
private List<Future<DLSN>> asyncWriteBulk(List<LogRecord> records) {
    final ArrayList<Future<DLSN>> results = new ArrayList<Future<DLSN>>(records.size());
    Iterator<LogRecord> iterator = records.iterator();
    while (iterator.hasNext()) {
        LogRecord record = iterator.next();
        Future<DLSN> future = asyncWrite(record, !iterator.hasNext());
        results.add(future);

        // Abort early if an individual write has already failed.
        Option<Try<DLSN>> result = future.poll();
        if (result.isDefined() && result.get().isThrow()) {
            break;
        }
    }
    if (records.size() > results.size()) {
        appendCancelledFutures(results, records.size() - results.size());
    }
    return results;
}
 

import com.twitter.util.Try; //导入依赖的package包/类
@Test
public void testGetEmptyClusters() {
  ByteBuffer key = ByteBuffer.wrap(KEY);
  TerrapinGetRequest request = prepareGetRequest();
  RequestOptions options = new RequestOptions();
  options.setSelectionPolicy(SelectionPolicy.PRIMARY_FIRST);
  request.setOptions(options);
  request.setClusterList(ImmutableList.copyOf(new String[]{}));

  TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of(
      key, new TerrapinSingleResponse().setValue(ByteBuffer.wrap("value".getBytes()))));
  when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn(
      Future.value(response));
  when(mockClient2.getManyNoRetries(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn(
      Future.value(response));

  Try<TerrapinSingleResponse> singleResponseTry = serviceIface.get(request).get(Duration.forever());
  assertTrue(singleResponseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.INVALID_REQUEST,
      ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode());

}
 

import com.twitter.util.Try; //导入依赖的package包/类
@Test
public void testMultiGetEmptyClusters() {
  TerrapinMultiGetRequest request = prepareMultiGetRequest();
  RequestOptions options = new RequestOptions();
  options.setSelectionPolicy(SelectionPolicy.PRIMARY_FIRST);
  request.setOptions(options);
  request.setClusterList(ImmutableList.copyOf(new String[]{}));

  TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of(
      ByteBuffer.wrap("key1".getBytes()),
      new TerrapinSingleResponse().setValue("value1".getBytes()),
      ByteBuffer.wrap("key2".getBytes()),
      new TerrapinSingleResponse().setErrorCode(TerrapinGetErrorCode.READ_ERROR)));
  Set<ByteBuffer> keys = Sets.newHashSet(ByteBuffer.wrap("key1".getBytes()),
      ByteBuffer.wrap("key2".getBytes()),
      ByteBuffer.wrap("key3".getBytes()));

  when(mockClient1.getMany(eq(FILESET), eq(keys))).thenReturn(Future.value(response));
  when(mockClient2.getManyNoRetries(eq(FILESET), eq(keys))).thenReturn(Future.value(response));

  Try<TerrapinResponse> returnResponseTry = serviceIface.multiGet(request).get(Duration.forever());
  assertTrue(returnResponseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.INVALID_REQUEST,
      ((TerrapinGetException)((Throw)returnResponseTry).e()).getErrorCode());
}
 

import com.twitter.util.Try; //导入依赖的package包/类
@Test
public void testMultiGetError() {
  TerrapinMultiGetRequest request = prepareMultiGetRequest();
  Set<ByteBuffer> keys = Sets.newHashSet(ByteBuffer.wrap("key1".getBytes()),
      ByteBuffer.wrap("key2".getBytes()),
      ByteBuffer.wrap("key3".getBytes()));

  when (mockClient1.getMany(eq(FILESET), eq(keys))).thenReturn(
      Future.<TerrapinResponse>exception(new TerrapinGetException("Failed",
          TerrapinGetErrorCode.FILE_SET_NOT_FOUND)));
  Try<TerrapinResponse> responseTry = serviceIface.multiGet(request).get(Duration.forever());
  assertTrue(responseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.FILE_SET_NOT_FOUND,
               ((TerrapinGetException)((Throw)responseTry).e()).getErrorCode());

  when(mockClient1.getMany(eq(FILESET), eq(keys))).thenThrow(
      new RuntimeException(new NullPointerException()));
  responseTry = serviceIface.multiGet(request).get(Duration.forever());
  assertTrue(responseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.OTHER,
               ((TerrapinGetException)((Throw)responseTry).e()).getErrorCode());
}
 

import com.twitter.util.Try; //导入依赖的package包/类
@Test
public void testGetErrorMultipleResources() throws Exception {
  TerrapinInternalGetRequest request = new TerrapinInternalGetRequest();
  MultiKey multiKey1 = new MultiKey().setResource("resource1").setPartition("1");
  multiKey1.addToKey(ByteBuffer.wrap("k1".getBytes()));
  MultiKey multiKey2 = new MultiKey().setResource("resource2").setPartition("1");
  multiKey2.addToKey(ByteBuffer.wrap("k2".getBytes()));
  request.addToKeyList(multiKey1);
  request.addToKeyList(multiKey2);

  Reader mockReader = mock(Reader.class);
  when(mockResourcePartitionMap.getReader(eq("resource1"), eq("1"))).thenReturn(mockReader);
  Try<TerrapinResponse> responseTry = serverImpl.get(request).get(Duration.forever());
  TerrapinGetException e = (TerrapinGetException)((Throw)responseTry).e();
  assertEquals(TerrapinGetErrorCode.INVALID_REQUEST, e.getErrorCode());
}
 

import com.twitter.util.Try; //导入依赖的package包/类
<T> void satisfyPromiseAsync(final Promise<T> promise, final Try<T> result) {
    scheduler.submit(new SafeRunnable() {
        @Override
        public void safeRun() {
            promise.update(result);
        }
    });
}
 

import com.twitter.util.Try; //导入依赖的package包/类
private boolean isDefiniteFailure(Try<DLSN> result) {
    boolean def = false;
    try {
        result.get();
    } catch (Exception ex) {
        if (ex instanceof OwnershipAcquireFailedException ||
            ex instanceof AlreadyClosedException ||
            ex instanceof LockingException) {
            def = true;
        }
    }
    return def;
}
 

import com.twitter.util.Try; //导入依赖的package包/类
private Future<List<Try<DLSN>>> asTryList(Future<List<Future<DLSN>>> futureList) {
    return futureList.flatMap(new AbstractFunction1<List<Future<DLSN>>, Future<List<Try<DLSN>>>>() {
        @Override
        public Future<List<Try<DLSN>>> apply(List<Future<DLSN>> results) {
            return Future$.MODULE$.collectToTry(results);
        }
    });
}
 

import com.twitter.util.Try; //导入依赖的package包/类
protected void setResponse(Response response) {
  Return<Response> responseTry = new Return(response);
  boolean isEmpty = result.updateIfEmpty(responseTry);
  if (!isEmpty) {
    Option<Try<Response>> resultTry = result.poll();
    logger.error("Result set multiple times. Value='{}', New='{}'", resultTry, responseTry);
  }
}
 

import com.twitter.util.Try; //导入依赖的package包/类
public Future<PinLaterDequeueResponse> dequeueJobs(final String source,
                                                   final PinLaterDequeueRequest request) {
  Future<PinLaterDequeueResponse> dequeueFuture;
  try {
    dequeueFuture = dequeueSemaphoreMap.get(request.getQueueName()).acquire().flatMap(
        new Function<Permit, Future<PinLaterDequeueResponse>>() {
          @Override
          public Future<PinLaterDequeueResponse> apply(final Permit permit) {
            return futurePool.apply(new ExceptionalFunction0<PinLaterDequeueResponse>() {
              @Override
              public PinLaterDequeueResponse applyE() throws Throwable {
                return dequeueJobsImpl(source, request, numAutoRetries);
              }
            }).respond(new Function<Try<PinLaterDequeueResponse>, BoxedUnit>() {
              @Override
              public BoxedUnit apply(Try<PinLaterDequeueResponse> responseTry) {
                permit.release();
                return BoxedUnit.UNIT;
              }
            });
          }
        });
  } catch (ExecutionException e) {
    // The dequeueSemaphoreMap's get() can in theory throw an ExecutionException, but we
    // never expect it in practice since our load method is simply new'ing up an AsyncSemaphore.
    dequeueFuture = Future.exception(e);
  }

  // Dequeue requests can contain ack requests as payloads. If so, we execute both in parallel.
  Future<Void> ackFuture = request.isSetJobAckRequest()
                           ? ackDequeuedJobsImpl(request.getJobAckRequest()) : Future.Void();

  return dequeueFuture.join(ackFuture).map(
      new Function<Tuple2<PinLaterDequeueResponse, Void>, PinLaterDequeueResponse>() {
        @Override
        public PinLaterDequeueResponse apply(Tuple2<PinLaterDequeueResponse, Void> tuple) {
          return tuple._1();
        }
      });
}
 

import com.twitter.util.Try; //导入依赖的package包/类
private void issueEnqueueRequests(PinLater.ServiceIface iface) throws InterruptedException {
  Preconditions.checkNotNull(queueName, "Queue was not specified.");
  final AtomicLong queriesIssued = new AtomicLong(0);
  final Semaphore permits = new Semaphore(concurrency);
  while (numQueries == -1 || queriesIssued.get() < numQueries) {
    final PinLaterEnqueueRequest request = new PinLaterEnqueueRequest();
    request.setQueueName(queueName);
    for (int i = 0; i < batchSize; i++) {
      PinLaterJob job = new PinLaterJob(ByteBuffer.wrap(
          new String("task_" + random.nextInt(Integer.MAX_VALUE)).getBytes()));
      job.setPriority(priority);
      request.addToJobs(job);
    }
    final long startTimeNanos = System.nanoTime();
    queriesIssued.incrementAndGet();
    permits.acquire();
    iface.enqueueJobs(REQUEST_CONTEXT, request).respond(
        new Function<Try<PinLaterEnqueueResponse>, BoxedUnit>() {
          @Override
          public BoxedUnit apply(Try<PinLaterEnqueueResponse> responseTry) {
            permits.release();
            statsLogger.requestComplete(
                Duration.fromNanoseconds(System.nanoTime() - startTimeNanos));
            if (responseTry.isThrow()) {
              LOG.info("Exception for request: " + request + " : " + ((Throw) responseTry).e());
            }
            return BoxedUnit.UNIT;
          }
        });
  }
  permits.acquire(concurrency);
  LOG.info("Enqueue queries issued: " + queriesIssued);
}
 

import com.twitter.util.Try; //导入依赖的package包/类
@Test
public void testGetClusterNotFound() {
  TerrapinGetRequest request = prepareGetRequest().setClusterList(ImmutableList.of(
      "random-cluster"));

  Try<TerrapinSingleResponse> singleResponseTry =
      serviceIface.get(request).get(Duration.forever());
  assertTrue(singleResponseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.CLUSTER_NOT_FOUND,
      ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode());
}
 

import com.twitter.util.Try; //导入依赖的package包/类
@Test
public void testGetError() {
  // Test the case where We get back an error set through an error code set in
  // TerrapinSingleResponse.
  ByteBuffer key = ByteBuffer.wrap(KEY);
  TerrapinGetRequest request = prepareGetRequest();

  TerrapinResponse response = new TerrapinResponse().setResponseMap(ImmutableMap.of(
      key, new TerrapinSingleResponse().setErrorCode(TerrapinGetErrorCode.OTHER)));
  when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn(
      Future.value(response));

  Try<TerrapinSingleResponse> singleResponseTry = serviceIface.get(request).get(
      Duration.forever());
  assertTrue(singleResponseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.OTHER,
      ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode());

  // Test the case where the call to the client library itself bails out due to a
  // legit error.
  when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenReturn(
          Future.<TerrapinResponse>exception(new TerrapinGetException("Failed.",
                  TerrapinGetErrorCode.FILE_SET_NOT_FOUND)));
  singleResponseTry = serviceIface.get(request).get(Duration.forever());
  assertTrue(singleResponseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.FILE_SET_NOT_FOUND,
      ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode());

  // Test the case where the call to the client library bails out due to a runtime
  // exception.
  when(mockClient1.getMany(eq(FILESET), eq(Sets.newHashSet(key)))).thenThrow(
      new RuntimeException(new NullPointerException()));
  singleResponseTry = serviceIface.get(request).get(Duration.forever());
  assertTrue(singleResponseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.OTHER,
      ((TerrapinGetException)((Throw)singleResponseTry).e()).getErrorCode());
}
 

import com.twitter.util.Try; //导入依赖的package包/类
@Test
public void testMultiGetClusterNotFound() {
  TerrapinMultiGetRequest request = prepareMultiGetRequest().setClusterList(ImmutableList.of(
      "random-cluster"));

  Try<TerrapinResponse> responseTry =
  serviceIface.multiGet(request).get(Duration.forever());
  assertTrue(responseTry.isThrow());
  assertEquals(TerrapinGetErrorCode.CLUSTER_NOT_FOUND,
               ((TerrapinGetException)((Throw)responseTry).e()).getErrorCode());
}
 

import com.twitter.util.Try; //导入依赖的package包/类
public static <T> Future<Try<T>> lifeToTry(Future<T> future) {
  return future.map(new Function<T, Try<T>>() {
    @Override
    public Try<T> apply(T o) {
      return new Return(o);
    }
  }).handle(new Function<Throwable, Try<T>>() {
    @Override
    public Try<T> apply(Throwable throwable) {
      return new Throw(throwable);
    }
  });
}
 

import com.twitter.util.Try; //导入依赖的package包/类
public GeneratorMaster() {
    redisCache = new RedisCache("localhost", 7000);

    String propertiesPath = "/videogenmaster.properties";
    PropertyPlaceholder propsHolder = new PropertyPlaceholder(propertiesPath);
    propsHolder.generatePropertyMap();
    Map<String, String> props = propsHolder.getPropertyMap();

    videoCacheUpdatePool = Executors.newFixedThreadPool(Integer.parseInt(props.get("maxThreads")));
    videoCacheUpdateTimer = new Timer();
    /*videoCacheUpdateTimer.schedule(new VideoCache(videoCacheUpdatePool, propsHolder, client), 
            5000, 5000);*/

    stats = new Statistics();

    RetryPolicy<Try<Nothing$>> retryPolicy = new SimpleRetryPolicy<Try<Nothing$>>() {

        @Override
        public Duration backoffAt(int retryCount) {
            if (retryCount > 3) {
                return Duration.fromSeconds(16);
            }
            return Duration.fromSeconds((int) Math.pow(2.0, retryCount));
        }

        @Override
        public boolean shouldRetry(Try<Nothing$> arg) {
            return true;
        }
    };

    ClientBuilder clientBuilder = ClientBuilder.get()
            .codec(com.twitter.finagle.http.Http.get())
            .retryPolicy(retryPolicy) // Retry forever, with exponential backoff <= 16
            .hostConnectionLimit(500)
            .hosts("localhost:" + SLAVE_PORT);
    client = ClientBuilder.safeBuild(clientBuilder);
    // client = Http.newService("localhost:8001");
}
 

import com.twitter.util.Try; //导入依赖的package包/类
private void issueDequeueAckRequests(final PinLater.ServiceIface iface)
    throws InterruptedException {
  Preconditions.checkNotNull(queueName, "Queue was not specified.");
  final AtomicLong queriesIssued = new AtomicLong(0);
  final Semaphore permits = new Semaphore(concurrency);
  while (numQueries == -1 || queriesIssued.get() < numQueries) {
    final PinLaterDequeueRequest request = new PinLaterDequeueRequest();
    request.setQueueName(queueName);
    request.setLimit(batchSize);
    final long startTimeNanos = System.nanoTime();
    queriesIssued.incrementAndGet();
    permits.acquire();
    iface.dequeueJobs(REQUEST_CONTEXT, request).flatMap(
        new Function<PinLaterDequeueResponse, Future<Void>>() {
          @Override
          public Future<Void> apply(PinLaterDequeueResponse response) {
            if (response.getJobsSize() == 0) {
              return Future.Void();
            }

            PinLaterJobAckRequest jobAckRequest = new PinLaterJobAckRequest(queueName);
            for (String job : response.getJobs().keySet()) {
              if (random.nextInt(100) < dequeueSuccessPercent) {
                jobAckRequest.addToJobsSucceeded(new PinLaterJobAckInfo(job));
              } else {
                jobAckRequest.addToJobsFailed(new PinLaterJobAckInfo(job));
              }
            }
            return iface.ackDequeuedJobs(REQUEST_CONTEXT, jobAckRequest);
          }
        }).respond(new Function<Try<Void>, BoxedUnit>() {
      @Override
      public BoxedUnit apply(Try<Void> voidTry) {
        permits.release();
        statsLogger.requestComplete(
            Duration.fromNanoseconds(System.nanoTime() - startTimeNanos));
        if (voidTry.isThrow()) {
          LOG.info("Exception for request: " + request + " : " + ((Throw) voidTry).e());
        }
        return BoxedUnit.UNIT;
      }
    });
  }
  permits.acquire(concurrency);
  LOG.info("Dequeue/ack queries issued: " + queriesIssued);
}
 

import com.twitter.util.Try; //导入依赖的package包/类
public GeneratorMasterZk() {
    redisCache = new RedisCache("localhost", 7000);

    String propertiesPath = "/videogenmaster.properties";
    PropertyPlaceholder propsHolder = new PropertyPlaceholder(propertiesPath);
    propsHolder.generatePropertyMap();
    Map<String, String> props = propsHolder.getPropertyMap();

    videoCacheUpdatePool = Executors.newFixedThreadPool(Integer.parseInt(props.get("maxThreads")));
    videoCacheUpdateTimer = new Timer();
    /*videoCacheUpdateTimer.schedule(new VideoCache(videoCacheUpdatePool, propsHolder, client), 
            5000, 5000);*/

    stats = new Statistics();
    
    InetSocketAddress zkServer = new InetSocketAddress("localhost", 2181);
    ZooKeeperClient zkClient = new ZooKeeperClient(Amount.of(5, Time.SECONDS), zkServer);
    ServerSetImpl serverSet = new ServerSetImpl(zkClient, "/videogen/slave");
    ZookeeperServerSetCluster cluster = new ZookeeperServerSetCluster(serverSet);
    
    RetryPolicy<Try<Nothing$>> retryPolicy = new SimpleRetryPolicy<Try<Nothing$>>() {
        @Override
        public Duration backoffAt(int retryCount) {
            if (retryCount > 3) {
                return Duration.fromSeconds(16);
            }
            return Duration.fromSeconds((int) Math.pow(2.0, retryCount));
        }

        @Override
        public boolean shouldRetry(Try<Nothing$> arg) {
            return true;
        }
    };

    ClientBuilder clientBuilder = ClientBuilder.get()
            .codec(com.twitter.finagle.http.Http.get())
            .retryPolicy(retryPolicy) // Retry forever, with exponential backoff <= 16
            .hostConnectionLimit(500)
            //.hosts("localhost:" + SLAVE_PORT);
            .cluster(cluster);
    client = ClientBuilder.safeBuild(clientBuilder);
    // client = Http.newService("localhost:8001");
}