Adding a pubsub publisher message transformer

[sduskis]

This will allow generic transformations of messages before a message is published.

This should help towards #4240

[codecov]

Codecov Report

Merging #4599 into master will decrease coverage by 0.32%.
The diff coverage is 100%.

@@             Coverage Diff              @@
##             master    #4599      +/-   ##
============================================
- Coverage     49.17%   48.84%   -0.33%     
  Complexity    21952    21952              
============================================
  Files          2078     2078              
  Lines        207459   207416      -43     
  Branches      24108    23416     -692     
============================================
- Hits         102015   101322     -693     
+ Misses        97270    97255      -15     
- Partials       8174     8839     +665

Impacted Files	Coverage Δ	Complexity Δ
...ain/java/com/google/cloud/pubsub/v1/Publisher.java	85.84% <100%> (+4.18%)	26 <0> (ø)	⬇️
...va/com/google/cloud/compute/v1/InstanceClient.java	48.96% <0%> (-6.44%)	127% <0%> (ø)
...ava/com/google/cloud/compute/v1/ProjectClient.java	50.69% <0%> (-6%)	55% <0%> (ø)
...ava/com/google/cloud/compute/v1/NetworkClient.java	49.26% <0%> (-5.89%)	35% <0%> (ø)
...ava/com/google/cloud/compute/v1/LicenseClient.java	50% <0%> (-5.84%)	31% <0%> (ø)
.../com/google/cloud/compute/v1/RegionDiskClient.java	48.55% <0%> (-5.8%)	35% <0%> (ø)
...va/com/google/cloud/compute/v1/SnapshotClient.java	49.58% <0%> (-5.79%)	31% <0%> (ø)
.../java/com/google/cloud/compute/v1/ImageClient.java	48.71% <0%> (-5.77%)	43% <0%> (ø)
.../google/cloud/compute/v1/TargetSslProxyClient.java	48.2% <0%> (-5.76%)	35% <0%> (ø)
.../google/cloud/compute/v1/SecurityPolicyClient.java	47.77% <0%> (-5.74%)	39% <0%> (ø)
... and 129 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 07e03b5...442f09b. Read the comment docs.

[dpcollins-google]

To me, this looks like it would likely be abused by users to implement a pipeline construct, which doesn't seem to be the intent. This would mean the order of application would have to be guaranteed and never changed to prevent breaking user code. What would you think about instead of having this be intrinsic, making Publisher an Abstract Class/Interface, and having the user wrap their publisher, similar to guava's ForwardingQueue? Something like

class CensusTrackedPublisher extends Publisher {
  // Could be implicit, here for clarity.
  CensusTrackedPublisher(Publisher delegate) {
     super(delegate);
  }

  @Override
  public ApiFuture<String> publish(PubsubMessage message) {
    return delegate().publish(censusTransform(message));
  }
}

[sduskis]

@dpcollins-google, it would seem to me that your suggestion would make the client backwards incompatible, which I don't think we can do.

[codecov-io]

Codecov Report

Merging #4599 into master will decrease coverage by 0.32%.
The diff coverage is 100%.

@@             Coverage Diff              @@
##             master    #4599      +/-   ##
============================================
- Coverage     49.17%   48.84%   -0.33%     
  Complexity    21952    21952              
============================================
  Files          2078     2078              
  Lines        207459   207416      -43     
  Branches      24108    23416     -692     
============================================
- Hits         102015   101322     -693     
+ Misses        97270    97255      -15     
- Partials       8174     8839     +665

Impacted Files	Coverage Δ	Complexity Δ
...ain/java/com/google/cloud/pubsub/v1/Publisher.java	85.84% <100%> (+4.18%)	26 <0> (ø)	⬇️
...va/com/google/cloud/compute/v1/InstanceClient.java	48.96% <0%> (-6.44%)	127% <0%> (ø)
...ava/com/google/cloud/compute/v1/ProjectClient.java	50.69% <0%> (-6%)	55% <0%> (ø)
...ava/com/google/cloud/compute/v1/NetworkClient.java	49.26% <0%> (-5.89%)	35% <0%> (ø)
...ava/com/google/cloud/compute/v1/LicenseClient.java	50% <0%> (-5.84%)	31% <0%> (ø)
.../com/google/cloud/compute/v1/RegionDiskClient.java	48.55% <0%> (-5.8%)	35% <0%> (ø)
...va/com/google/cloud/compute/v1/SnapshotClient.java	49.58% <0%> (-5.79%)	31% <0%> (ø)
.../java/com/google/cloud/compute/v1/ImageClient.java	48.71% <0%> (-5.77%)	43% <0%> (ø)
.../google/cloud/compute/v1/TargetSslProxyClient.java	48.2% <0%> (-5.76%)	35% <0%> (ø)
.../google/cloud/compute/v1/SecurityPolicyClient.java	47.77% <0%> (-5.74%)	39% <0%> (ø)
... and 129 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 07e03b5...442f09b. Read the comment docs.

[dpcollins-google]

@sduskis in what way? Publisher only has a private constructor and can only be instantiated through the Builder. If Publisher were instead a public abstract class with one abstract method (Publish) and Builder.build() instead returned a {package-private} class PublisherImpl extends Publisher, I don't think any user code would have to be changed. There would be 0 api changes relative to the current state. Other languages (python, javascript, go) are dynamic enough that this would already be possible without the explicit interface, and you could maintain parity with no changes to other languages.

[sduskis]

@dpcollins-google, there are public static methods on Publisher which we would not be able to support in Java 7.

[dhoard]

@sduskis This type of functionality seems like it should be above the Pub/Sub client since it's specific to the application.

class CensusMessagePublisher {

  private Publisher publisher;

  public CensusMessagePublisher(Publisher publisher) {
     this.publisher = publisher;  
  }

  @Override
  public ApiFuture<String> publish(CensusMessage message) {
    return this.publisher.publish(transform(message));
  }

  private PubsubMessage transform(CensusMessage message) {
      // transform the CensusMessage to a PubsubMessage
      return transformedMessage;
   }
}

You could also pass in a transform function in the constructor and use it ... but again still seems application specific.

[sduskis]

@dhoard, thanks for the guidance. FWIW, this change was in context of #4240 which is OpenCensus / PubSub. A review of #4240 would also be greatly appreciated.