com.google.common.io.Closer#create ( )源码实例Demo

@Test(dependsOnMethods = {"testSetAndGet"})
public void testSerDe()
    throws IOException {
  Closer closer = Closer.create();
  try {
    ByteArrayOutputStream baos = closer.register(new ByteArrayOutputStream());
    DataOutputStream dos = closer.register(new DataOutputStream(baos));
    this.jobState.write(dos);

    ByteArrayInputStream bais = closer.register((new ByteArrayInputStream(baos.toByteArray())));
    DataInputStream dis = closer.register((new DataInputStream(bais)));
    JobState newJobState = new JobState();
    newJobState.readFields(dis);
    doAsserts(newJobState, true, false);
  } catch (Throwable t) {
    throw closer.rethrow(t);
  } finally {
    closer.close();
  }
}
 

@Override
public void encode(final ByteBuf out, final Object message) throws IOException {
    Closer closer = Closer.create();
    try {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        closer.register(byteArrayOutputStream);
        KryoSerialize kryoSerialization = new KryoSerialize(pool);
        kryoSerialization.serialize(byteArrayOutputStream, message);
        byte[] body = byteArrayOutputStream.toByteArray();
        int dataLength = body.length;
        out.writeInt(dataLength);
        out.writeBytes(body);
    } finally {
        closer.close();
    }
}
 

@Test
public void testTarGzCopy() throws Exception {

  Closer closer = Closer.create();
  try {
    JobLauncher jobLauncher = closer.register(JobLauncherFactory.newJobLauncher(gobblinProps, jobProps));
    jobLauncher.launchJob(null);

    String file1Path =
        gobblinProps.getProperty(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR) + "/LogData/sub1/sub2/text1.txt";
    String file2Path =
        gobblinProps.getProperty(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR) + "/LogData/sub1/sub2/text2.txt";

    FileSystem fs = FileSystem.getLocal(new Configuration());

    Assert.assertEquals(IOUtils.toString(closer.register(fs.open(new Path(file1Path)))), "text1");
    Assert.assertEquals(IOUtils.toString(closer.register(fs.open(new Path(file2Path)))), "text2");

  } finally {
    closer.close();
  }
}
 

@AfterClass
public void cleanup()
{
    if (servers != null) {
        try (Closer closer = Closer.create()) {
            for (DriftServer server : servers) {
                closer.register(() -> server.shutdown());
            }
        }
        catch (IOException e) {
            throw new RuntimeException(e);
        }

        servers = null;
    }
}
 

/**
 * Test when a test with the matching suffix is skipped.
 * @param jobProps job properties
 * @param skippedTaskSuffix the suffix for the task that is skipped
 */
public void runTestWithSkippedTask(Properties jobProps, String skippedTaskSuffix) throws Exception {
  String jobName = jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY);
  String jobId = JobLauncherUtils.newJobId(jobName).toString();
  jobProps.setProperty(ConfigurationKeys.JOB_ID_KEY, jobId);
  jobProps.setProperty(ConfigurationKeys.PUBLISH_DATA_AT_JOB_LEVEL, Boolean.FALSE.toString());
  jobProps.setProperty(ConfigurationKeys.JOB_COMMIT_POLICY_KEY, "successful");
  jobProps.setProperty(ConfigurationKeys.MAX_TASK_RETRIES_KEY, "0");

  Closer closer = Closer.create();
  try {
    JobLauncher jobLauncher = closer.register(JobLauncherFactory.newJobLauncher(this.launcherProps, jobProps));
    jobLauncher.launchJob(null);
  } finally {
    closer.close();
  }

  List<JobState.DatasetState> datasetStateList =
      this.datasetStateStore.getAll(jobName, sanitizeJobNameForDatasetStore(jobId) + ".jst");
  JobState jobState = datasetStateList.get(0);

  Assert.assertEquals(jobState.getState(), JobState.RunningState.COMMITTED);
  // one task is skipped out of 4
  Assert.assertEquals(jobState.getCompletedTasks(), 3);
  for (TaskState taskState : jobState.getTaskStates()) {
    if (taskState.getTaskId().endsWith(skippedTaskSuffix)) {
      Assert.assertEquals(taskState.getWorkingState(), WorkUnitState.WorkingState.PENDING);
    } else {
      Assert.assertEquals(taskState.getWorkingState(), WorkUnitState.WorkingState.COMMITTED);
      Assert.assertEquals(taskState.getPropAsLong(ConfigurationKeys.WRITER_RECORDS_WRITTEN),
          TestExtractor.TOTAL_RECORDS);
    }
  }
}
 

@BeforeTestWithContext
public void setUp()
{
    closer = Closer.create();
    executor = newSingleThreadExecutor(); // single thread is enough, it schedules the query to cancel
    closer.register(executor::shutdownNow);
    queryCanceller = closer.register(new QueryCanceller(serverAddress));
}
 

@Override
@SuppressWarnings("unchecked")
public T get(String storeName, String tableName, String stateId) throws IOException {
  Path tablePath = new Path(new Path(this.storeRootDir, storeName), tableName);
  if (!this.fs.exists(tablePath)) {
    return null;
  }

  Closer closer = Closer.create();
  try {
    @SuppressWarnings("deprecation")
    GobblinSequenceFileReader reader = closer.register(new GobblinSequenceFileReader(this.fs, tablePath, this.conf));
    try {
      Text key = new Text();
      T state = this.stateClass.newInstance();
      while (reader.next(key)) {
        state = (T)reader.getCurrentValue(state);
        if (key.toString().equals(stateId)) {
          state.setId(stateId);
          return state;
        }
      }
    } catch (Exception e) {
      throw new IOException("failure retrieving state from storeName " + storeName + " tableName " + tableName, e);
    }
  } catch (Throwable t) {
    throw closer.rethrow(t);
  } finally {
    closer.close();
  }

  return null;
}
 

static void defineClassesInBootstrapClassLoader(Collection<LazyDefinedClass> lazyDefinedClasses,
        Instrumentation instrumentation, File generatedJarFile) throws IOException {
    Closer closer = Closer.create();
    try {
        FileOutputStream out = closer.register(new FileOutputStream(generatedJarFile));
        JarOutputStream jarOut = closer.register(new JarOutputStream(out));
        generate(lazyDefinedClasses, jarOut);
    } catch (Throwable t) {
        throw closer.rethrow(t);
    } finally {
        closer.close();
    }
    instrumentation.appendToBootstrapClassLoaderSearch(new JarFile(generatedJarFile));
    // appendToBootstrapClassLoaderSearch() line above does not add to the bootstrap resource
    // search path, only to the bootstrap class search path (this is different from
    // appendToSystemClassLoaderSearch() which adds to both the system resource search path and
    // the system class search path)
    //
    // adding the generated jar file to the bootstrap resource search path is probably needed
    // more generally, but it is at least needed to support jboss 4.2.0 - 4.2.3 because
    // org.jboss.mx.loading.LoadMgr3.beginLoadTask() checks that the class loader has the class
    // as a resource before loading it, so without adding the generated jar file to the
    // bootstrap resource search path, jboss ends up throwing ClassNotFoundException for the
    // glowroot generated classes that have been added to the bootstrap class loader search path
    // (see issue #101 for more info on this particular jboss issue)
    appendToBootstrapResourcePath(generatedJarFile);
}
 

@Override
public void close()
{
    try (Closer closer = Closer.create()) {
        closer.register(master::stop);
        tServers.forEach(tabletServer -> closer.register(tabletServer::stop));
    }
    catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

/**
 * Run a job.
 *
 * <p>
 *   This method runs the job immediately without going through the Quartz scheduler.
 *   This is particularly useful for testing.
 * </p>
 *
 * <p>
 *   This method does what {@link #runJob(Properties, JobListener)} does, and additionally it allows
 *   the caller to pass in a {@link JobLauncher} instance used to launch the job to run.
 * </p>
 *
 * @param jobProps Job configuration properties
 * @param jobListener {@link JobListener} used for callback, can be <em>null</em> if no callback is needed.
 * @param jobLauncher a {@link JobLauncher} object used to launch the job to run
 * @return If current job is a stop-early job based on {@link Source#isEarlyStopped()}
 * @throws JobException when there is anything wrong with running the job
 */
public boolean runJob(Properties jobProps, JobListener jobListener, JobLauncher jobLauncher)
    throws JobException {
  Preconditions.checkArgument(jobProps.containsKey(ConfigurationKeys.JOB_NAME_KEY),
      "A job must have a job name specified by job.name");
  String jobName = jobProps.getProperty(ConfigurationKeys.JOB_NAME_KEY);

  // Check if the job has been disabled
  boolean disabled = Boolean.valueOf(jobProps.getProperty(ConfigurationKeys.JOB_DISABLED_KEY, "false"));
  if (disabled) {
    LOG.info("Skipping disabled job " + jobName);
    return false;
  }

  // Launch the job
  try (Closer closer = Closer.create()) {
    closer.register(jobLauncher).launchJob(jobListener);
    boolean runOnce = Boolean.valueOf(jobProps.getProperty(ConfigurationKeys.JOB_RUN_ONCE_KEY, "false"));
    boolean isEarlyStopped = jobLauncher.isEarlyStopped();
    if (!isEarlyStopped && runOnce && this.scheduledJobs.containsKey(jobName)) {
      this.scheduler.getScheduler().deleteJob(this.scheduledJobs.remove(jobName));
    }

    return isEarlyStopped;

  } catch (Throwable t) {
    throw new JobException("Failed to launch and run job " + jobName, t);
  }
}
 

@Override
public void close()
        throws IOException
{
    try (Closer closer = Closer.create()) {
        closer.register(super::close);
        if (credentialsProvider instanceof Closeable) {
            closer.register((Closeable) credentialsProvider);
        }
        closer.register(s3::shutdown);
    }
}
 

public static void zip(InputStream is, OutputStream os) throws IOException {
    Closer closer = Closer.create();
    closer.register(is);
    try {
        GZIPOutputStream zos = new GZIPOutputStream(os);
        closer.register(zos);
        ByteStreams.copy(is, zos);
    } catch (IOException ioe) {
        throw closer.rethrow(ioe);
    } finally {
        closer.close();
    }
}
 

/**
 * Save the result to a file.
 *
 * @param file the file
 * @return this object
 * @throws java.io.IOException on I/O error
 */
public HttpRequest saveContent(File file) throws IOException {

    try (Closer closer = Closer.create()) {
        FileOutputStream fos = closer.register(new FileOutputStream(file));
        BufferedOutputStream bos = closer.register(new BufferedOutputStream(fos));

        saveContent(bos);
    }

    return this;
}
 

private static <T> T processAndClose(InputStream in, InputStreamProcessor<T> processor)
        throws IOException {
    Closer closer = Closer.create();
    try {
        closer.register(in);
        return processor.process(in);
    } catch (Throwable t) {
        throw closer.rethrow(t);
    } finally {
        closer.close();
    }
}
 

@Test
public void testMissingJobLockType_ResultsIn_FileBasedJobLock() throws JobLockException, IOException {
  Closer closer = Closer.create();
  try {
    Properties properties = new Properties();
    properties.setProperty(ConfigurationKeys.FS_URI_KEY, "file:///");
    properties.setProperty(FileBasedJobLock.JOB_LOCK_DIR, "JobLockFactoryTest");
    properties.setProperty(ConfigurationKeys.JOB_NAME_KEY, "JobLockFactoryTest-" + System.currentTimeMillis());
    properties.setProperty(ConfigurationKeys.JOB_LOCK_TYPE, FileBasedJobLock.class.getName());
    JobLock jobLock = closer.register(LegacyJobLockFactoryManager.getJobLock(properties, new JobLockEventListener()));
    MatcherAssert.assertThat(jobLock, Matchers.instanceOf(FileBasedJobLock.class));
  } finally {
    closer.close();
  }
}
 

@Override
public Serializable readFrom(Class<Serializable> type, Type genericType, Annotation[] annotations,
        MediaType mediaType, MultivaluedMap<String, String> httpHeaders, InputStream entityStream)
        throws IOException, WebApplicationException {
    Closer closer = Closer.create();
    try {
        entityStream = closer.register(entityStream);
        return CharStreams.toString(new InputStreamReader(entityStream));
    } catch (IOException ioe) {
        throw closer.rethrow(ioe);
    } finally {
        closer.close();
    }
}
 

@Override
public void close() throws IOException {
    if (!_openIterators.isEmpty()) {
        try {
            // Use a closer to cleanly close all iterators even if one throws an exception on close
            Closer closer = Closer.create();
            for (StashRowIterator iterator : _openIterators) {
                closer.register(iterator);
            }
            closer.close();
        } finally {
            _openIterators.clear();
        }
    }
}
 

public DistributedIndexServer(Configuration configuration, ZooKeeper zookeeper, ClusterStatus clusterStatus,
    BlurFilterCache filterCache, BlockCacheDirectoryFactory blockCacheDirectoryFactory,
    DistributedLayoutFactory distributedLayoutFactory, String cluster, String nodeName, long safeModeDelay,
    int shardOpenerThreadCount, int maxMergeThreads, int internalSearchThreads,
    int minimumNumberOfNodesBeforeExitingSafeMode, Timer hdfsKeyValueTimer, Timer indexImporterTimer,
    long smallMergeThreshold, Timer indexBulkTimer, ThriftCache thriftCache,
    SequentialReadControl sequentialReadControl, Timer indexIdleWriterTimer, long maxWriterIdle)
    throws KeeperException, InterruptedException {
  super(clusterStatus, configuration, nodeName, cluster);
  _indexIdleWriterTimer = indexIdleWriterTimer;
  _maxWriterIdle = maxWriterIdle;
  _sequentialReadControl = sequentialReadControl;
  _indexImporterTimer = indexImporterTimer;
  _indexBulkTimer = indexBulkTimer;
  _hdfsKeyValueTimer = hdfsKeyValueTimer;
  _minimumNumberOfNodes = minimumNumberOfNodesBeforeExitingSafeMode;
  _running.set(true);
  _closer = Closer.create();
  _shardOpenerThreadCount = shardOpenerThreadCount;
  _zookeeper = zookeeper;
  _filterCache = filterCache;
  _safeModeDelay = safeModeDelay;
  _internalSearchThreads = internalSearchThreads;
  _blockCacheDirectoryFactory = blockCacheDirectoryFactory;
  _distributedLayoutFactory = distributedLayoutFactory;
  _thriftCache = thriftCache;

  _closer.register(_shardStateManager);

  BlurUtil.setupZookeeper(_zookeeper, _cluster);
  _openerService = Executors.newThreadPool("shard-opener", _shardOpenerThreadCount);
  _searchExecutor = Executors.newThreadPool("internal-search", _internalSearchThreads);

  _closer.register(CloseableExecutorService.close(_openerService));
  _closer.register(CloseableExecutorService.close(_searchExecutor));

  // @TODO allow for configuration of these
  _mergeScheduler = _closer.register(new SharedMergeScheduler(maxMergeThreads, smallMergeThreshold));

  _indexCloser = _closer.register(new BlurIndexCloser());
  _timerCacheFlush = setupFlushCacheTimer();
  _timerCacheFlush.start();

  String onlineShardsPath = ZookeeperPathConstants.getOnlineShardsPath(_cluster);
  String safemodePath = ZookeeperPathConstants.getSafemodePath(_cluster);

  // Set the registerNode timeout value to zk sessionTimeout + {4} seconds
  int registerNodeTimeOut = _zookeeper.getSessionTimeout() / 1000 + 4;

  SafeMode safeMode = new SafeMode(_zookeeper, safemodePath, onlineShardsPath, TimeUnit.MILLISECONDS, _safeModeDelay,
      TimeUnit.SECONDS, registerNodeTimeOut, _minimumNumberOfNodes);
  safeMode.registerNode(getNodeName(), BlurUtil.getVersion().getBytes());

  _timerTableWarmer = setupTableWarmer();
  _timerTableWarmer.start();
  _watchOnlineShards = watchForShardServerChanges();
  _clusterStatus.registerActionOnTableStateChange(new Action() {
    @Override
    public void action() {
      synchronized (_warmupLock) {
        _warmupLock.notifyAll();
      }
    }
  });
  _clusterStatus.registerActionOnTableStateChange(new Action() {
    @Override
    public void action() {
      synchronized (_cleanupLock) {
        _cleanupLock.notifyAll();
      }
    }
  });
}
 

public ClassIndex(String name, ImmutableSet<Path> jarFiles, Predicate<Path> isDirect)
    throws IOException {
  this.name = name;
  this.closer = Closer.create();
  classIndex = buildClassIndex(jarFiles, closer, isDirect);
}
 

@VisibleForTesting
static void splitIntoShards(Options options) throws IOException {
  checkArgument(
      !options.minimalMainDex || options.mainDexListFile != null,
      "--minimal-main-dex not allowed without --main-dex-list");

  if (!Files.exists(options.outputDirectory)) {
    Files.createDirectories(options.outputDirectory);
  }

  ImmutableSet<String> classesInMainDex =
      options.mainDexListFile != null
          ? ImmutableSet.copyOf(Files.readAllLines(options.mainDexListFile, UTF_8))
          : null;
  ImmutableSet<String> expected =
      options.inclusionFilterJar != null ? expectedEntries(options.inclusionFilterJar) : null;
  try (Closer closer = Closer.create();
      DexFileSplitter out =
          new DexFileSplitter(options.outputDirectory, options.maxNumberOfIdxPerDex)) {
    // 1. Scan inputs in order and keep first occurrence of each class, keeping all zips open.
    // We don't process anything yet so we can shard in sorted order, which is what dx would do
    // if presented with a single jar containing all the given inputs.
    // TODO(kmb): Abandon alphabetic sorting to process each input fully before moving on (still
    // requires scanning inputs twice for main dex list).
    Predicate<ZipEntry> inclusionFilter = ZipEntryPredicates.suffixes(".dex", ".class");
    if (expected != null) {
      inclusionFilter = inclusionFilter.and(e -> expected.contains(e.getName()));
    }
    LinkedHashMap<String, ZipFile> deduped = new LinkedHashMap<>();
    for (Path inputArchive : options.inputArchives) {
      ZipFile zip = closer.register(new ZipFile(inputArchive.toFile()));
      zip.stream()
          .filter(inclusionFilter)
          .forEach(e -> deduped.putIfAbsent(e.getName(), zip));
    }
    ImmutableList<Map.Entry<String, ZipFile>> files =
        deduped
            .entrySet()
            .stream()
            .sorted(Comparator.comparing(e -> e.getKey(), ZipEntryComparator::compareClassNames))
            .collect(ImmutableList.toImmutableList());

    // 2. Process each class in desired order, rolling from shard to shard as needed.
    if (classesInMainDex == null || classesInMainDex.isEmpty()) {
      out.processDexFiles(files, Predicates.alwaysTrue());
    } else {
      checkArgument(classesInMainDex.stream().noneMatch(s -> s.startsWith("j$/")),
          "%s lists classes in package 'j$', which can't be included in classes.dex and can "
              + "cause runtime errors. Please avoid needing these classes in the main dex file.",
          options.mainDexListFile);
      // To honor --main_dex_list make two passes:
      // 1. process only the classes listed in the given file
      // 2. process the remaining files
      Predicate<String> mainDexFilter = ZipEntryPredicates.classFileNameFilter(classesInMainDex);
      out.processDexFiles(files, mainDexFilter);
      // Fail if main_dex_list is too big, following dx's example
      checkState(out.shardsWritten() == 0, "Too many classes listed in main dex list file "
          + "%s, main dex capacity exceeded", options.mainDexListFile);
      if (options.minimalMainDex) {
        out.nextShard(); // Start new .dex file if requested
      }
      out.processDexFiles(files, mainDexFilter.negate());
    }
  }
}