@Test public void putReturnsMd5() throws Exception { String checksum1 = storage.put("key1", 10, contents("0123456789")); String checksum2 = storage.put("key2", 5, contents("01234")); assertThat(checksum1, either(is(md5hex("0123456789"))).or(is(""))); assertThat(checksum2, either(is(md5hex("01234"))).or(is(""))); }
@Override public String put(String object, long contentLength, UploadStreamProvider payload) throws IOException { checkArgument(object != null, "object is null"); BlobInfo blobInfo = BlobInfo.newBuilder(bucket, object).build(); try { return uploadRetryExecutor() .onRetry((exception, retryCount, retryLimit, retryWait) -> { logger.warn("Retrying uploading file bucket " + bucket + " object " + object + " error: " + exception); }) .retryIf((exception) -> { if (exception instanceof IOException || exception instanceof InterruptedException) { return false; } return true; }) .runInterruptible(() -> { try (InputStream in = payload.open()) { try (WriteChannel writer = storage.writer(blobInfo)) { byte[] buffer = new byte[1024]; int limit; while ((limit = in.read(buffer)) >= 0) { try { writer.write(ByteBuffer.wrap(buffer, 0, limit)); } catch (Exception ex) { ex.printStackTrace(); } } } return storage.get(bucket, object).getMd5ToHexString(); } }); } catch (InterruptedException ex) { throw Throwables.propagate(ex); } catch (RetryGiveupException ex) { Throwable cause = ex.getCause(); Throwables.propagateIfInstanceOf(cause, IOException.class); throw Throwables.propagate(cause); } }
GCSStorage implements io.digdag.spi.Storage { @Override public String put(String object, long contentLength, UploadStreamProvider payload) throws IOException { checkArgument(object != null, "object is null"); BlobInfo blobInfo = BlobInfo.newBuilder(bucket, object).build(); try { return uploadRetryExecutor() .onRetry((exception, retryCount, retryLimit, retryWait) -> { logger.warn("Retrying uploading file bucket " + bucket + " object " + object + " error: " + exception); }) .retryIf((exception) -> { if (exception instanceof IOException || exception instanceof InterruptedException) { return false; } return true; }) .runInterruptible(() -> { try (InputStream in = payload.open()) { try (WriteChannel writer = storage.writer(blobInfo)) { byte[] buffer = new byte[1024]; int limit; while ((limit = in.read(buffer)) >= 0) { try { writer.write(ByteBuffer.wrap(buffer, 0, limit)); } catch (Exception ex) { ex.printStackTrace(); } } } return storage.get(bucket, object).getMd5ToHexString(); } }); } catch (InterruptedException ex) { throw Throwables.propagate(ex); } catch (RetryGiveupException ex) { Throwable cause = ex.getCause(); Throwables.propagateIfInstanceOf(cause, IOException.class); throw Throwables.propagate(cause); } } }
GCSStorage implements io.digdag.spi.Storage { @Override public String put(String object, long contentLength, UploadStreamProvider payload) throws IOException { checkArgument(object != null, "object is null"); BlobInfo blobInfo = BlobInfo.newBuilder(bucket, object).build(); try { return uploadRetryExecutor() .onRetry((exception, retryCount, retryLimit, retryWait) -> { logger.warn("Retrying uploading file bucket " + bucket + " object " + object + " error: " + exception); }) .retryIf((exception) -> { if (exception instanceof IOException || exception instanceof InterruptedException) { return false; } return true; }) .runInterruptible(() -> { try (InputStream in = payload.open()) { try (WriteChannel writer = storage.writer(blobInfo)) { byte[] buffer = new byte[1024]; int limit; while ((limit = in.read(buffer)) >= 0) { try { writer.write(ByteBuffer.wrap(buffer, 0, limit)); } catch (Exception ex) { ex.printStackTrace(); } } } return storage.get(bucket, object).getMd5ToHexString(); } }); } catch (InterruptedException ex) { throw Throwables.propagate(ex); } catch (RetryGiveupException ex) { Throwable cause = ex.getCause(); Throwables.propagateIfInstanceOf(cause, IOException.class); throw Throwables.propagate(cause); } } GCSStorage(final Config config, com.google.cloud.storage.Storage storage, String bucket); }
GCSStorage implements io.digdag.spi.Storage { @Override public String put(String object, long contentLength, UploadStreamProvider payload) throws IOException { checkArgument(object != null, "object is null"); BlobInfo blobInfo = BlobInfo.newBuilder(bucket, object).build(); try { return uploadRetryExecutor() .onRetry((exception, retryCount, retryLimit, retryWait) -> { logger.warn("Retrying uploading file bucket " + bucket + " object " + object + " error: " + exception); }) .retryIf((exception) -> { if (exception instanceof IOException || exception instanceof InterruptedException) { return false; } return true; }) .runInterruptible(() -> { try (InputStream in = payload.open()) { try (WriteChannel writer = storage.writer(blobInfo)) { byte[] buffer = new byte[1024]; int limit; while ((limit = in.read(buffer)) >= 0) { try { writer.write(ByteBuffer.wrap(buffer, 0, limit)); } catch (Exception ex) { ex.printStackTrace(); } } } return storage.get(bucket, object).getMd5ToHexString(); } }); } catch (InterruptedException ex) { throw Throwables.propagate(ex); } catch (RetryGiveupException ex) { Throwable cause = ex.getCause(); Throwables.propagateIfInstanceOf(cause, IOException.class); throw Throwables.propagate(cause); } } GCSStorage(final Config config, com.google.cloud.storage.Storage storage, String bucket); @Override StorageObject open(String object); @Override String put(String object, long contentLength,
            UploadStreamProvider payload); @Override void list(String objectPrefix, FileListing callback); @Override Optional<DirectDownloadHandle> getDirectDownloadHandle(String object); @Override Optional<DirectUploadHandle> getDirectUploadHandle(String object); }
GCSStorage implements io.digdag.spi.Storage { @Override public String put(String object, long contentLength, UploadStreamProvider payload) throws IOException { checkArgument(object != null, "object is null"); BlobInfo blobInfo = BlobInfo.newBuilder(bucket, object).build(); try { return uploadRetryExecutor() .onRetry((exception, retryCount, retryLimit, retryWait) -> { logger.warn("Retrying uploading file bucket " + bucket + " object " + object + " error: " + exception); }) .retryIf((exception) -> { if (exception instanceof IOException || exception instanceof InterruptedException) { return false; } return true; }) .runInterruptible(() -> { try (InputStream in = payload.open()) { try (WriteChannel writer = storage.writer(blobInfo)) { byte[] buffer = new byte[1024]; int limit; while ((limit = in.read(buffer)) >= 0) { try { writer.write(ByteBuffer.wrap(buffer, 0, limit)); } catch (Exception ex) { ex.printStackTrace(); } } } return storage.get(bucket, object).getMd5ToHexString(); } }); } catch (InterruptedException ex) { throw Throwables.propagate(ex); } catch (RetryGiveupException ex) { Throwable cause = ex.getCause(); Throwables.propagateIfInstanceOf(cause, IOException.class); throw Throwables.propagate(cause); } } GCSStorage(final Config config, com.google.cloud.storage.Storage storage, String bucket); @Override StorageObject open(String object); @Override String put(String object, long contentLength,
            UploadStreamProvider payload); @Override void list(String objectPrefix, FileListing callback); @Override Optional<DirectDownloadHandle> getDirectDownloadHandle(String object); @Override Optional<DirectUploadHandle> getDirectUploadHandle(String object);  }