beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From lc...@apache.org
Subject [1/3] incubator-beam git commit: Update PubsubGrpcClient to not take in mocks as they are not needed anymore Update PubsubGrpcClientTest to use an inprocess server to handle requests/responses for testing. Add/remove required dependencies found missing b
Date Thu, 22 Sep 2016 21:43:10 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master 8245f9b4b -> 4872bde8f


Update PubsubGrpcClient to not take in mocks as they are not needed anymore
Update PubsubGrpcClientTest to use an inprocess server to handle requests/responses for testing.
Add/remove required dependencies found missing by dep-analyzer plugin during upgrade to protobuf
3.0.0 and grpc 1.0.1


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a447d130
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a447d130
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a447d130

Branch: refs/heads/master
Commit: a447d130172ec7c270fc9f1d16a2f7404898461e
Parents: f93ca9c
Author: Luke Cwik <lcwik@google.com>
Authored: Wed Sep 21 19:49:10 2016 -0700
Committer: Luke Cwik <lcwik@google.com>
Committed: Thu Sep 22 14:37:00 2016 -0700

----------------------------------------------------------------------
 pom.xml                                         |  24 +++++
 runners/google-cloud-dataflow-java/pom.xml      |   6 ++
 sdks/java/core/pom.xml                          |  17 +++
 .../apache/beam/sdk/util/PubsubGrpcClient.java  |  34 ++----
 .../beam/sdk/util/PubsubGrpcClientTest.java     | 108 ++++++++++++-------
 sdks/java/io/google-cloud-platform/pom.xml      |   5 +
 6 files changed, 132 insertions(+), 62 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a447d130/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index 70373ec..cc93bb9 100644
--- a/pom.xml
+++ b/pom.xml
@@ -321,11 +321,35 @@
 
       <dependency>
         <groupId>io.grpc</groupId>
+        <artifactId>grpc-protobuf</artifactId>
+        <version>${grpc.version}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>io.grpc</groupId>
+        <artifactId>grpc-protobuf-lite</artifactId>
+        <version>${grpc.version}</version>
+      </dependency>
+
+      <dependency>
+        <groupId>com.google.protobuf</groupId>
+        <artifactId>protobuf-lite</artifactId>
+        <version>3.0.1</version>
+      </dependency>
+
+      <dependency>
+        <groupId>io.grpc</groupId>
         <artifactId>grpc-netty</artifactId>
         <version>${grpc.version}</version>
       </dependency>
 
       <dependency>
+        <groupId>io.grpc</groupId>
+        <artifactId>grpc-stub</artifactId>
+        <version>${grpc.version}</version>
+      </dependency>
+
+      <dependency>
         <groupId>com.google.api-client</groupId>
         <artifactId>google-api-client</artifactId>
         <version>${google-clients.version}</version>

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a447d130/runners/google-cloud-dataflow-java/pom.xml
----------------------------------------------------------------------
diff --git a/runners/google-cloud-dataflow-java/pom.xml b/runners/google-cloud-dataflow-java/pom.xml
index bf66f38..acf6cce 100644
--- a/runners/google-cloud-dataflow-java/pom.xml
+++ b/runners/google-cloud-dataflow-java/pom.xml
@@ -321,6 +321,12 @@
     </dependency>
 
     <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-lite</artifactId>
+      <scope>runtime</scope>
+    </dependency>
+
+    <dependency>
       <groupId>com.google.code.findbugs</groupId>
       <artifactId>annotations</artifactId>
     </dependency>

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a447d130/sdks/java/core/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/core/pom.xml b/sdks/java/core/pom.xml
index 1cf7ba5..c4d3e64 100644
--- a/sdks/java/core/pom.xml
+++ b/sdks/java/core/pom.xml
@@ -308,6 +308,11 @@
       <artifactId>grpc-netty</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-stub</artifactId>
+    </dependency>
+
     <!-- grpc-all does not obey IWYU, so we need to exclude from compile
          scope and depend on it at runtime. -->
     <dependency>
@@ -317,6 +322,18 @@
     </dependency>
 
     <dependency>
+      <groupId>io.grpc</groupId>
+      <artifactId>grpc-protobuf-lite</artifactId>
+      <scope>runtime</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-lite</artifactId>
+      <scope>runtime</scope>
+    </dependency>
+
+    <dependency>
       <groupId>com.google.auth</groupId>
       <artifactId>google-auth-library-oauth2-http</artifactId>
       <version>0.4.0</version>

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a447d130/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PubsubGrpcClient.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PubsubGrpcClient.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PubsubGrpcClient.java
index 988b90f..02152ba 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PubsubGrpcClient.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PubsubGrpcClient.java
@@ -100,9 +100,7 @@ public class PubsubGrpcClient extends PubsubClient {
                                   idLabel,
                                   DEFAULT_TIMEOUT_S,
                                   channel,
-                                  credentials,
-                                  null /* publisher stub */,
-                                  null /* subscriber stub */);
+                                  credentials);
     }
 
     @Override
@@ -159,16 +157,12 @@ public class PubsubGrpcClient extends PubsubClient {
       @Nullable String idLabel,
       int timeoutSec,
       ManagedChannel publisherChannel,
-      GoogleCredentials credentials,
-      PublisherGrpc.PublisherBlockingStub cachedPublisherStub,
-      SubscriberGrpc.SubscriberBlockingStub cachedSubscriberStub) {
+      GoogleCredentials credentials) {
     this.timestampLabel = timestampLabel;
     this.idLabel = idLabel;
     this.timeoutSec = timeoutSec;
     this.publisherChannel = publisherChannel;
     this.credentials = credentials;
-    this.cachedPublisherStub = cachedPublisherStub;
-    this.cachedSubscriberStub = cachedSubscriberStub;
   }
 
   /**
@@ -189,13 +183,11 @@ public class PubsubGrpcClient extends PubsubClient {
     this.publisherChannel = null;
     // Gracefully shutdown the channel.
     publisherChannel.shutdown();
-    if (timeoutSec > 0) {
-      try {
-        publisherChannel.awaitTermination(timeoutSec, TimeUnit.SECONDS);
-      } catch (InterruptedException e) {
-        // Ignore.
-        Thread.currentThread().interrupt();
-      }
+    try {
+      publisherChannel.awaitTermination(timeoutSec, TimeUnit.SECONDS);
+    } catch (InterruptedException e) {
+      // Ignore.
+      Thread.currentThread().interrupt();
     }
   }
 
@@ -216,11 +208,7 @@ public class PubsubGrpcClient extends PubsubClient {
     if (cachedPublisherStub == null) {
       cachedPublisherStub = PublisherGrpc.newBlockingStub(newChannel());
     }
-    if (timeoutSec > 0) {
-      return cachedPublisherStub.withDeadlineAfter(timeoutSec, TimeUnit.SECONDS);
-    } else {
-      return cachedPublisherStub;
-    }
+    return cachedPublisherStub.withDeadlineAfter(timeoutSec, TimeUnit.SECONDS);
   }
 
   /**
@@ -230,11 +218,7 @@ public class PubsubGrpcClient extends PubsubClient {
     if (cachedSubscriberStub == null) {
       cachedSubscriberStub = SubscriberGrpc.newBlockingStub(newChannel());
     }
-    if (timeoutSec > 0) {
-      return cachedSubscriberStub.withDeadlineAfter(timeoutSec, TimeUnit.SECONDS);
-    } else {
-      return cachedSubscriberStub;
-    }
+    return cachedSubscriberStub.withDeadlineAfter(timeoutSec, TimeUnit.SECONDS);
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a447d130/sdks/java/core/src/test/java/org/apache/beam/sdk/util/PubsubGrpcClientTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/PubsubGrpcClientTest.java
b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/PubsubGrpcClientTest.java
index b36b934..cbdf5da 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/PubsubGrpcClientTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/PubsubGrpcClientTest.java
@@ -23,19 +23,26 @@ import static org.junit.Assert.assertEquals;
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Iterables;
 import com.google.protobuf.ByteString;
 import com.google.protobuf.Timestamp;
 import com.google.pubsub.v1.PublishRequest;
 import com.google.pubsub.v1.PublishResponse;
-import com.google.pubsub.v1.PublisherGrpc;
+import com.google.pubsub.v1.PublisherGrpc.PublisherImplBase;
 import com.google.pubsub.v1.PubsubMessage;
 import com.google.pubsub.v1.PullRequest;
 import com.google.pubsub.v1.PullResponse;
 import com.google.pubsub.v1.ReceivedMessage;
-import com.google.pubsub.v1.SubscriberGrpc;
+import com.google.pubsub.v1.SubscriberGrpc.SubscriberImplBase;
 import io.grpc.ManagedChannel;
+import io.grpc.Server;
+import io.grpc.inprocess.InProcessChannelBuilder;
+import io.grpc.inprocess.InProcessServerBuilder;
+import io.grpc.stub.StreamObserver;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.ThreadLocalRandom;
 import org.apache.beam.sdk.util.PubsubClient.IncomingMessage;
 import org.apache.beam.sdk.util.PubsubClient.OutgoingMessage;
 import org.apache.beam.sdk.util.PubsubClient.SubscriptionPath;
@@ -52,12 +59,11 @@ import org.mockito.Mockito;
  */
 @RunWith(JUnit4.class)
 public class PubsubGrpcClientTest {
-  private ManagedChannel mockChannel;
+  private ManagedChannel inProcessChannel;
   private GoogleCredentials mockCredentials;
-  private PublisherGrpc.PublisherBlockingStub mockPublisherStub;
-  private SubscriberGrpc.SubscriberBlockingStub mockSubscriberStub;
 
   private PubsubClient client;
+  private String channelName;
 
   private static final TopicPath TOPIC = PubsubClient.topicPathFromName("testProject", "testTopic");
   private static final SubscriptionPath SUBSCRIPTION =
@@ -73,31 +79,24 @@ public class PubsubGrpcClientTest {
   private static final String ACK_ID = "testAckId";
 
   @Before
-  public void setup() throws IOException {
-    mockChannel = Mockito.mock(ManagedChannel.class);
+  public void setup() {
+    channelName = String.format("%s-%s",
+        PubsubGrpcClientTest.class.getName(), ThreadLocalRandom.current().nextInt());
+    inProcessChannel = InProcessChannelBuilder.forName(channelName).directExecutor().build();
     mockCredentials = Mockito.mock(GoogleCredentials.class);
-    mockPublisherStub =
-        Mockito.mock(PublisherGrpc.PublisherBlockingStub.class, Mockito.RETURNS_DEEP_STUBS);
-    mockSubscriberStub =
-        Mockito.mock(SubscriberGrpc.SubscriberBlockingStub.class, Mockito.RETURNS_DEEP_STUBS);
-    client = new PubsubGrpcClient(TIMESTAMP_LABEL, ID_LABEL, 0, mockChannel,
-                                  mockCredentials, mockPublisherStub, mockSubscriberStub);
+    client = new PubsubGrpcClient(TIMESTAMP_LABEL, ID_LABEL, 10, inProcessChannel, mockCredentials);
   }
 
   @After
   public void teardown() throws IOException {
     client.close();
-    client = null;
-    mockChannel = null;
-    mockCredentials = null;
-    mockPublisherStub = null;
-    mockSubscriberStub = null;
+    inProcessChannel.shutdownNow();
   }
 
   @Test
   public void pullOneMessage() throws IOException {
     String expectedSubscription = SUBSCRIPTION.getPath();
-    PullRequest expectedRequest =
+    final PullRequest expectedRequest =
         PullRequest.newBuilder()
                    .setSubscription(expectedSubscription)
                    .setReturnImmediately(true)
@@ -123,20 +122,37 @@ public class PubsubGrpcClientTest {
                        .setMessage(expectedPubsubMessage)
                        .setAckId(ACK_ID)
                        .build();
-    PullResponse expectedResponse =
+    final PullResponse response =
         PullResponse.newBuilder()
                     .addAllReceivedMessages(ImmutableList.of(expectedReceivedMessage))
                     .build();
-    Mockito.when(mockSubscriberStub.pull(expectedRequest))
-           .thenReturn(expectedResponse);
-    List<IncomingMessage> acutalMessages = client.pull(REQ_TIME, SUBSCRIPTION, 10,
true);
-    assertEquals(1, acutalMessages.size());
-    IncomingMessage actualMessage = acutalMessages.get(0);
-    assertEquals(ACK_ID, actualMessage.ackId);
-    assertEquals(DATA, new String(actualMessage.elementBytes));
-    assertEquals(RECORD_ID, actualMessage.recordId);
-    assertEquals(REQ_TIME, actualMessage.requestTimeMsSinceEpoch);
-    assertEquals(MESSAGE_TIME, actualMessage.timestampMsSinceEpoch);
+
+    final List<PullRequest> requestsReceived = new ArrayList<>();
+    SubscriberImplBase subscriberImplBase = new SubscriberImplBase() {
+      @Override
+      public void pull(PullRequest request, StreamObserver<PullResponse> responseObserver)
{
+        requestsReceived.add(request);
+        responseObserver.onNext(response);
+        responseObserver.onCompleted();
+      }
+    };
+    Server server = InProcessServerBuilder.forName(channelName)
+        .addService(subscriberImplBase)
+        .build()
+        .start();
+    try {
+      List<IncomingMessage> acutalMessages = client.pull(REQ_TIME, SUBSCRIPTION, 10,
true);
+      assertEquals(1, acutalMessages.size());
+      IncomingMessage actualMessage = acutalMessages.get(0);
+      assertEquals(ACK_ID, actualMessage.ackId);
+      assertEquals(DATA, new String(actualMessage.elementBytes));
+      assertEquals(RECORD_ID, actualMessage.recordId);
+      assertEquals(REQ_TIME, actualMessage.requestTimeMsSinceEpoch);
+      assertEquals(MESSAGE_TIME, actualMessage.timestampMsSinceEpoch);
+      assertEquals(expectedRequest, Iterables.getOnlyElement(requestsReceived));
+    } finally {
+      server.shutdownNow();
+    }
   }
 
   @Test
@@ -149,20 +165,38 @@ public class PubsubGrpcClientTest {
                          ImmutableMap.of(TIMESTAMP_LABEL, String.valueOf(MESSAGE_TIME),
                                          ID_LABEL, RECORD_ID))
                      .build();
-    PublishRequest expectedRequest =
+    final PublishRequest expectedRequest =
         PublishRequest.newBuilder()
                       .setTopic(expectedTopic)
                       .addAllMessages(
                           ImmutableList.of(expectedPubsubMessage))
                       .build();
-    PublishResponse expectedResponse =
+    final PublishResponse response =
         PublishResponse.newBuilder()
                        .addAllMessageIds(ImmutableList.of(MESSAGE_ID))
                        .build();
-    Mockito.when(mockPublisherStub.publish(expectedRequest))
-           .thenReturn(expectedResponse);
-    OutgoingMessage actualMessage = new OutgoingMessage(DATA.getBytes(), MESSAGE_TIME, RECORD_ID);
-    int n = client.publish(TOPIC, ImmutableList.of(actualMessage));
-    assertEquals(1, n);
+
+    final List<PublishRequest> requestsReceived = new ArrayList<>();
+    PublisherImplBase publisherImplBase = new PublisherImplBase() {
+      @Override
+      public void publish(
+          PublishRequest request, StreamObserver<PublishResponse> responseObserver)
{
+        requestsReceived.add(request);
+        responseObserver.onNext(response);
+        responseObserver.onCompleted();
+      }
+    };
+    Server server = InProcessServerBuilder.forName(channelName)
+        .addService(publisherImplBase)
+        .build()
+        .start();
+    try {
+      OutgoingMessage actualMessage = new OutgoingMessage(DATA.getBytes(), MESSAGE_TIME,
RECORD_ID);
+      int n = client.publish(TOPIC, ImmutableList.of(actualMessage));
+      assertEquals(1, n);
+      assertEquals(expectedRequest, Iterables.getOnlyElement(requestsReceived));
+    } finally {
+      server.shutdownNow();
+    }
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a447d130/sdks/java/io/google-cloud-platform/pom.xml
----------------------------------------------------------------------
diff --git a/sdks/java/io/google-cloud-platform/pom.xml b/sdks/java/io/google-cloud-platform/pom.xml
index a0f3fef..573c457 100644
--- a/sdks/java/io/google-cloud-platform/pom.xml
+++ b/sdks/java/io/google-cloud-platform/pom.xml
@@ -124,6 +124,11 @@
     </dependency>
 
     <dependency>
+      <groupId>com.google.protobuf</groupId>
+      <artifactId>protobuf-lite</artifactId>
+    </dependency>
+
+    <dependency>
       <groupId>io.grpc</groupId>
       <artifactId>grpc-core</artifactId>
     </dependency>


Mime
View raw message