1.0.9 (#9)

* Update gRPC version
* Add support for streaming commands

README.md

@@ -30,7 +30,7 @@ Regenerate the python stubs whenever changes are made to the *src/main/proto/ros
 
 ## RosBridge Usage
 
-The *ros-grpc-bridge* jar files are published to [Jitpack.io](https://jitpack.io/#athenian-robotics/ros-grpc-bridge/1.0.8).
+The *ros-grpc-bridge* jar files are published to [Jitpack.io](https://jitpack.io/#athenian-robotics/ros-grpc-bridge/1.0.9).
 
 ### Gradle
 
@@ -48,7 +48,7 @@ Add the JitPack repository and dependecy to your root *build.gradle*:
 
 ```groovy
 	dependencies {
-	        compile 'com.github.athenian-robotics:ros-grpc-bridge:1.0.8'
+	        compile 'com.github.athenian-robotics:ros-grpc-bridge:1.0.9'
 	}
 ```
 
@@ -69,7 +69,7 @@ Add the JitPack repository and dependecy to your *pom.xml*:
 	<dependency>
 	    <groupId>com.github.athenian-robotics</groupId>
 	    <artifactId>ros-grpc-bridge</artifactId>
-	    <version>1.0.8</version>
+	    <version>1.0.9</version>
 	</dependency>
 ```
 

build.gradle

@@ -21,7 +21,7 @@ repositories {
 sourceCompatibility = 1.8
 targetCompatibility = 1.8
 
-def grpcVersion = '1.9.0'
+def grpcVersion = '1.9.1'
 
 dependencies {
     testCompile group: 'junit', name: 'junit', version: '4.12'

pom.xml

@@ -22,14 +22,14 @@
 
     <groupId>org.athenian</groupId>
     <artifactId>ros-grpc-bridge</artifactId>
-    <version>1.0.8-SNAPSHOT</version>
+    <version>1.0.9-SNAPSHOT</version>
 
 
     <properties>
         <jcommander.version>1.72</jcommander.version>
         <guava.version>19.0</guava.version>
 
-        <grpc.version>1.9.0</grpc.version>
+        <grpc.version>1.9.1</grpc.version>
         <grpc.plugin.version>1.9.0</grpc.plugin.version>
         <protobuf.version>3.5.1</protobuf.version>
         <protobuf.plugin.version>0.5.1</protobuf.plugin.version>

src/main/java/org/athenian/RosBridgeClient.java

@@ -16,6 +16,7 @@
 
 package org.athenian;
 
+import com.google.protobuf.Empty;
 import com.google.protobuf.StringValue;
 import io.grpc.ManagedChannel;
 import io.grpc.Status;
@@ -25,6 +26,7 @@
 import org.athenian.common.Utils;
 import org.athenian.core.RosBridgeClientOptions;
 import org.athenian.core.TwistValueStream;
+import org.athenian.grpc.CommandValue;
 import org.athenian.grpc.EncoderValue;
 import org.athenian.grpc.RosBridgeServiceGrpc.RosBridgeServiceBlockingStub;
 import org.athenian.grpc.RosBridgeServiceGrpc.RosBridgeServiceStub;
@@ -126,16 +128,32 @@ public static void main(final String[] argv)
       }
       Utils.sleepForSecs(1);
 
-      final Iterator<EncoderValue> iter = client.encoderValues("wheel1");
-      while (iter.hasNext()) {
-        EncoderValue encoderValue = iter.next();
+      final Iterator<EncoderValue> encoder_iter = client.encoderValues("wheel1");
+      while (encoder_iter.hasNext()) {
+        EncoderValue encoderValue = encoder_iter.next();
         logger.info("Read encoder value: " + encoderValue.getValue());
       }
 
-      final CountDownLatch completedLatch =
+      final CountDownLatch encoderLatch =
           client.encoderValues("wheel1",
                                encoderValue -> logger.info("Read encoder value: " + encoderValue.getValue()));
-      completedLatch.await();
+      encoderLatch.await();
+      Utils.sleepForSecs(1);
+
+      final Iterator<CommandValue> iter = client.commandValues();
+      while (iter.hasNext()) {
+        CommandValue commandValue = iter.next();
+        logger.info("Read command: " + commandValue.getCommand());
+        logger.info("Read command arg: " + commandValue.getCommandArg());
+      }
+
+      final CountDownLatch commandLatch =
+          client.commandValues(
+              commandValue -> {
+                logger.info("Read command: " + commandValue.getCommand());
+                logger.info("Read command arg: " + commandValue.getCommandArg());
+              });
+      commandLatch.await();
       Utils.sleepForSecs(1);
     }
   }
@@ -177,6 +195,36 @@ public void onCompleted() {
     return completedLatch;
   }
 
+  public Iterator<CommandValue> commandValues() {
+    return this.getBlockingStub().readCommandValues(Empty.getDefaultInstance());
+  }
+
+  public CountDownLatch commandValues(final Consumer<CommandValue> onMessageAction) {
+    final CountDownLatch completedLatch = new CountDownLatch(1);
+    final StreamObserver<CommandValue> observer = new StreamObserver<CommandValue>() {
+      @Override
+      public void onNext(CommandValue value) {
+        if (onMessageAction != null)
+          onMessageAction.accept(value);
+      }
+
+      @Override
+      public void onError(Throwable t) {
+        final Status status = Status.fromThrowable(t);
+        if (status != Status.CANCELLED)
+          logger.info("Error in asyncEncoderValues(): {}", status);
+        this.onCompleted();
+      }
+
+      @Override
+      public void onCompleted() {
+        completedLatch.countDown();
+      }
+    };
+    this.getNonBlockingStub().readCommandValues(Empty.getDefaultInstance(), observer);
+    return completedLatch;
+  }
+
   public TwistValueStream newTwistValueStream() {
     return new TwistValueStream(this.getNonBlockingStub());
   }

src/main/java/org/athenian/core/RosBridgeServiceImpl.java

@@ -23,6 +23,7 @@
 import io.grpc.stub.StreamObserver;
 import org.athenian.RosBridgeClient;
 import org.athenian.RosBridgeServer;
+import org.athenian.grpc.CommandValue;
 import org.athenian.grpc.EncoderValue;
 import org.athenian.grpc.RosBridgeServiceGrpc;
 import org.athenian.grpc.TwistValue;
@@ -89,4 +90,15 @@ public void readEncoderValues(StringValue request, StreamObserver<EncoderValue>
       observer.onNext(EncoderValue.newBuilder().setValue(i).build());
     observer.onCompleted();
   }
+
+  @Override
+  public void readCommandValues(Empty request, StreamObserver<CommandValue> observer) {
+    logger.info("Returning command");
+    for (int i = 0; i < RosBridgeClient.COUNT; i++)
+      observer.onNext(CommandValue.newBuilder()
+                                  .setCommand(String.format("Command %d", i))
+                                  .setCommandArg(String.format("Command Arg %d", i))
+                                  .build());
+    observer.onCompleted();
+  }
 }

src/main/proto/rosbridge_service.proto

@@ -20,10 +20,17 @@ message EncoderValue {
     double value = 1;
 }
 
+message CommandValue {
+    string command = 1;
+    string command_arg = 2;
+}
+
 service RosBridgeService {
     rpc writeTwistValue (TwistValue) returns (google.protobuf.Empty);
 
     rpc streamTwistValues (stream TwistValue) returns (google.protobuf.Empty);
 
     rpc readEncoderValues (google.protobuf.StringValue) returns (stream EncoderValue);
+
+    rpc readCommandValues (google.protobuf.Empty) returns (stream CommandValue);
 }

src/main/python/rosbridge_client.py

@@ -1,5 +1,6 @@
 import grpc
 import logging
+from google.protobuf.empty_pb2 import *
 from google.protobuf.wrappers_pb2 import *
 
 from stubs.rosbridge_service_pb2 import RosBridgeServiceStub
@@ -35,9 +36,16 @@ def stream_twist(self, iter_val):
             self.__log_error("Failed to reach gRPC server at {0} [{1}]".format(self.__hostname, e))
             raise e
 
-    def read_encoder(self, encoder_name):
+    def read_encoders(self, encoder_name):
         try:
-            return self.__stub.readEncoderValues(StringValue(name=encoder_name))
+            return self.__stub.readEncoderValues(StringValue(value=encoder_name))
+        except BaseException as e:
+            self.__log_error("Failed to reach gRPC server at {0} [{1}]".format(self.__hostname, e))
+            raise e
+
+    def read_commands(self):
+        try:
+            return self.__stub.readCommandValues(Empty())
         except BaseException as e:
             self.__log_error("Failed to reach gRPC server at {0} [{1}]".format(self.__hostname, e))
             raise e

src/main/python/rosbridge_test.py

@@ -33,10 +33,15 @@ def twist_gen(cnt):
 
     time.sleep(2)
 
-    # Read streming encoder values
-    for data in bridge.read_encoder("wheel2"):
+    # Read streaming encoder values
+    for data in bridge.read_encoders("wheel2"):
         logger.info("Read encoder value: " + str(data.value))
 
+    # Read streaming command values
+    for data in bridge.read_commands():
+        logger.info("Read command value: " + str(data.command))
+        logger.info("Read command value: " + str(data.command_arg))
+
 
 if __name__ == "__main__":
     main()