@Test public void testGzipCompress() { String text = StringUtils.repeat("Lorem ipsum dolor sit amet ", 100); byte[] compressedBytes = CompressPayload.compress(text.getBytes(StandardCharsets.UTF_8), Compression.GZIP); assertThat(ArrayUtils.toObject(compressedBytes), Matchers.arrayWithSize(68)); }
@VisibleForTesting PubsubMessage compress(PubsubMessage message) { message = PubsubConstraints.ensureNonNull(message); byte[] compressedBytes = compress(message.getPayload(), compression.get()); if (compressedBytes.length > maxCompressedBytes) { byte[] truncated = Arrays.copyOfRange(message.getPayload(), 0, maxCompressedBytes); truncationCounter.inc(); compressedBytes = compress(truncated, compression.get()); } return new PubsubMessage(compressedBytes, message.getAttributeMap()); }
CompressPayload extends PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> { @VisibleForTesting PubsubMessage compress(PubsubMessage message) { message = PubsubConstraints.ensureNonNull(message); byte[] compressedBytes = compress(message.getPayload(), compression.get()); if (compressedBytes.length > maxCompressedBytes) { byte[] truncated = Arrays.copyOfRange(message.getPayload(), 0, maxCompressedBytes); truncationCounter.inc(); compressedBytes = compress(truncated, compression.get()); } return new PubsubMessage(compressedBytes, message.getAttributeMap()); } }
CompressPayload extends PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> { @VisibleForTesting PubsubMessage compress(PubsubMessage message) { message = PubsubConstraints.ensureNonNull(message); byte[] compressedBytes = compress(message.getPayload(), compression.get()); if (compressedBytes.length > maxCompressedBytes) { byte[] truncated = Arrays.copyOfRange(message.getPayload(), 0, maxCompressedBytes); truncationCounter.inc(); compressedBytes = compress(truncated, compression.get()); } return new PubsubMessage(compressedBytes, message.getAttributeMap()); } private  CompressPayload(ValueProvider<Compression> compression, int maxCompressedBytes); }
CompressPayload extends PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> { @VisibleForTesting PubsubMessage compress(PubsubMessage message) { message = PubsubConstraints.ensureNonNull(message); byte[] compressedBytes = compress(message.getPayload(), compression.get()); if (compressedBytes.length > maxCompressedBytes) { byte[] truncated = Arrays.copyOfRange(message.getPayload(), 0, maxCompressedBytes); truncationCounter.inc(); compressedBytes = compress(truncated, compression.get()); } return new PubsubMessage(compressedBytes, message.getAttributeMap()); } private  CompressPayload(ValueProvider<Compression> compression, int maxCompressedBytes); static CompressPayload of(ValueProvider<Compression> compression); CompressPayload withMaxCompressedBytes(int maxBytes); @Override PCollection<PubsubMessage> expand(PCollection<? extends PubsubMessage> input); }
CompressPayload extends PTransform<PCollection<? extends PubsubMessage>, PCollection<PubsubMessage>> { @VisibleForTesting PubsubMessage compress(PubsubMessage message) { message = PubsubConstraints.ensureNonNull(message); byte[] compressedBytes = compress(message.getPayload(), compression.get()); if (compressedBytes.length > maxCompressedBytes) { byte[] truncated = Arrays.copyOfRange(message.getPayload(), 0, maxCompressedBytes); truncationCounter.inc(); compressedBytes = compress(truncated, compression.get()); } return new PubsubMessage(compressedBytes, message.getAttributeMap()); } private  CompressPayload(ValueProvider<Compression> compression, int maxCompressedBytes); static CompressPayload of(ValueProvider<Compression> compression); CompressPayload withMaxCompressedBytes(int maxBytes); @Override PCollection<PubsubMessage> expand(PCollection<? extends PubsubMessage> input);  }