Streaming messages between client and server services using gRPC

  1<?xml version="1.0" encoding="UTF-8"?>
  2<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd" xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
  3
  4    <modelVersion>4.0.0</modelVersion>
  5    <groupId>io.openliberty.guides</groupId>
  6    <artifactId>guide-grpc-intro-system</artifactId>
  7    <version>1.0-SNAPSHOT</version>
  8    <packaging>war</packaging>
  9
 10    <properties>
 11        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
 12        <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 13        <maven.compiler.source>11</maven.compiler.source>
 14        <maven.compiler.target>11</maven.compiler.target>
 15        <!-- Liberty configuration -->
 16        <liberty.var.http.port>9081</liberty.var.http.port>
 17        <liberty.var.https.port>9444</liberty.var.https.port>
 18    </properties>
 19    <dependencies>
 20        <!-- Provided dependencies -->
 21        <dependency>
 22            <groupId>jakarta.platform</groupId>
 23            <artifactId>jakarta.jakartaee-api</artifactId>
 24            <version>10.0.0</version>
 25            <scope>provided</scope>
 26        </dependency>
 27        <dependency>
 28            <groupId>org.eclipse.microprofile</groupId>
 29            <artifactId>microprofile</artifactId>
 30            <version>6.1</version>
 31            <type>pom</type>
 32            <scope>provided</scope>
 33        </dependency>
 34
 35        <dependency>
 36           <groupId>io.openliberty.guides</groupId>
 37           <artifactId>systemproto</artifactId>
 38           <version>1.0-SNAPSHOT</version>
 39        </dependency>
 40
 41        <!-- Java utility classes -->
 42        <dependency>
 43            <groupId>org.apache.commons</groupId>
 44            <artifactId>commons-lang3</artifactId>
 45            <version>3.17.0</version>
 46        </dependency>
 47
 48        <!-- grpc compile dependencies; provided by Liberty at runtime -->
 49        <dependency>
 50            <groupId>io.grpc</groupId>
 51            <artifactId>grpc-protobuf</artifactId>
 52            <version>1.68.2</version>
 53            <scope>provided</scope>
 54        </dependency>
 55        <dependency>
 56            <groupId>io.grpc</groupId>
 57            <artifactId>grpc-stub</artifactId>
 58            <version>1.68.2</version>
 59            <scope>provided</scope>
 60        </dependency>
 61
 62        <!-- For tests -->
 63        <dependency>
 64            <groupId>org.junit.jupiter</groupId>
 65            <artifactId>junit-jupiter</artifactId>
 66            <version>5.11.3</version>
 67            <scope>test</scope>
 68        </dependency>
 69        <!-- tag::grpc-testing[] -->
 70        <dependency>
 71            <groupId>io.grpc</groupId>
 72            <artifactId>grpc-testing</artifactId>
 73            <version>1.68.2</version>
 74            <scope>test</scope>
 75        </dependency>
 76        <!-- end::grpc-testing[] -->
 77        <!-- tag::mockito-core[] -->
 78        <dependency>
 79            <groupId>org.mockito</groupId>
 80            <artifactId>mockito-core</artifactId>
 81            <version>5.14.2</version>
 82            <scope>test</scope>
 83        </dependency>
 84        <!-- end::mockito-core[] -->
 85    </dependencies>
 86    
 87    <build>
 88        <finalName>${project.artifactId}</finalName>
 89        <plugins>
 90            <plugin>
 91                <groupId>org.apache.maven.plugins</groupId>
 92                <artifactId>maven-war-plugin</artifactId>
 93                <version>3.4.0</version>
 94            </plugin>
 95            <plugin>
 96                <groupId>io.openliberty.tools</groupId>
 97                <artifactId>liberty-maven-plugin</artifactId>
 98                <version>3.11.1</version>
 99            </plugin>
100            <plugin>
101                <groupId>org.apache.maven.plugins</groupId>
102                <artifactId>maven-surefire-plugin</artifactId>
103                <version>3.5.2</version>
104            </plugin>       
105        </plugins>
106    </build>
107</project>

  1// tag::copyright[]
  2/*******************************************************************************
  3 * Copyright (c) 2023 IBM Corporation and others.
  4 * All rights reserved. This program and the accompanying materials
  5 * are made available under the terms of the Eclipse Public License 2.0
  6 * which accompanies this distribution, and is available at
  7 * http://www.eclipse.org/legal/epl-2.0/
  8 *
  9 * SPDX-License-Identifier: EPL-2.0
 10 *******************************************************************************/
 11// end::copyright[]
 12package io.openliberty.guides.system;
 13
 14import static org.junit.jupiter.api.Assertions.assertEquals;
 15import static org.junit.jupiter.api.Assertions.assertTrue;
 16import static org.junit.jupiter.api.Assertions.fail;
 17import static org.mockito.ArgumentMatchers.any;
 18import static org.mockito.Mockito.mock;
 19import static org.mockito.Mockito.never;
 20import static org.mockito.Mockito.timeout;
 21import static org.mockito.Mockito.verify;
 22
 23import java.util.ArrayList;
 24import java.util.List;
 25import java.util.concurrent.CountDownLatch;
 26import java.util.concurrent.TimeUnit;
 27import java.util.stream.Collectors;
 28
 29import org.junit.jupiter.api.AfterAll;
 30import org.junit.jupiter.api.BeforeAll;
 31import org.junit.jupiter.api.Test;
 32import org.mockito.ArgumentCaptor;
 33
 34import io.grpc.ManagedChannel;
 35import io.grpc.Server;
 36import io.grpc.inprocess.InProcessChannelBuilder;
 37import io.grpc.inprocess.InProcessServerBuilder;
 38import io.grpc.stub.StreamObserver;
 39import io.openliberty.guides.systemproto.SystemProperties;
 40import io.openliberty.guides.systemproto.SystemProperty;
 41import io.openliberty.guides.systemproto.SystemPropertyName;
 42import io.openliberty.guides.systemproto.SystemPropertyPrefix;
 43import io.openliberty.guides.systemproto.SystemPropertyValue;
 44import io.openliberty.guides.systemproto.SystemServiceGrpc;
 45import io.openliberty.guides.systemproto.SystemServiceGrpc.SystemServiceBlockingStub;
 46import io.openliberty.guides.systemproto.SystemServiceGrpc.SystemServiceStub;
 47
 48public class SystemServiceTest {
 49
 50    private static final String SERVER_NAME = "system";
 51
 52    private static Server inProcessServer;
 53    private static ManagedChannel inProcessChannel;
 54    private static SystemServiceBlockingStub blockingStub;
 55    private static SystemServiceStub asyncStub;
 56
 57    @BeforeAll
 58    public static void setUp() throws Exception {
 59        // tag::inProcessServer[]
 60        inProcessServer = InProcessServerBuilder.forName(SERVER_NAME)
 61                              .addService(new SystemService())
 62                              .directExecutor()
 63                              .build();
 64        inProcessServer.start();
 65        // end::inProcessServer[]
 66        // tag::inProcessChannel[]
 67        inProcessChannel = InProcessChannelBuilder.forName(SERVER_NAME)
 68                               .directExecutor()
 69                               .build();
 70        // end::inProcessChannel[]
 71        // tag::blockingStub[]
 72        blockingStub = SystemServiceGrpc.newBlockingStub(inProcessChannel);
 73        // end::blockingStub[]
 74        // tag::asyncStub[]
 75        asyncStub = SystemServiceGrpc.newStub(inProcessChannel);
 76        // end::asyncStub[]
 77    }
 78
 79    // tag::tearDown[]
 80    @AfterAll
 81    public static void tearDown() {
 82        inProcessChannel.shutdownNow();
 83        inProcessServer.shutdownNow();
 84    }
 85    // end::tearDown[]
 86
 87    @Test
 88    // tag::testGetProperty[]
 89    public void testGetProperty() throws Exception {
 90        SystemPropertyName request = SystemPropertyName.newBuilder()
 91                                         .setPropertyName("os.name")
 92                                         .build();
 93        SystemPropertyValue response = blockingStub.getProperty(request);
 94        assertEquals(System.getProperty("os.name"), response.getPropertyValue());
 95    }
 96    // end::testGetProperty[]
 97
 98    @Test
 99    // tag::testGetServerStreamingProperties[]
100    public void testGetServerStreamingProperties() throws Exception {
101
102        SystemPropertyPrefix request = SystemPropertyPrefix.newBuilder()
103                                           .setPropertyPrefix("os.")
104                                           .build();
105        final CountDownLatch countDown = new CountDownLatch(1);
106        List<SystemProperty> properties = new ArrayList<SystemProperty>();
107        StreamObserver<SystemProperty> responseObserver =
108            new StreamObserver<SystemProperty>() {
109                @Override
110                public void onNext(SystemProperty property) {
111                    properties.add(property);
112                }
113
114                @Override
115                public void onError(Throwable t) {
116                    fail(t.getMessage());
117                }
118
119                @Override
120                public void onCompleted() {
121                    countDown.countDown();
122                }
123            };
124
125        asyncStub.getServerStreamingProperties(request, responseObserver);
126        assertTrue(countDown.await(10, TimeUnit.SECONDS));
127
128        for (SystemProperty property : properties) {
129            String propertName = property.getPropertyName();
130            assertEquals(System.getProperty(propertName),
131                property.getPropertyValue(), propertName + " is incorrect");
132        }
133    }
134    // end::testGetServerStreamingProperties[]
135
136    @Test
137    // tag::testGetClientStreamingProperties[]
138    public void testGetClientStreamingProperties() {
139
140        @SuppressWarnings("unchecked")
141        StreamObserver<SystemProperties> responseObserver =
142            (StreamObserver<SystemProperties>) mock(StreamObserver.class);
143        ArgumentCaptor<SystemProperties> systemPropertiesCaptor =
144            ArgumentCaptor.forClass(SystemProperties.class);
145
146        StreamObserver<SystemPropertyName> requestObserver =
147            asyncStub.getClientStreamingProperties(responseObserver);
148        List<String> keys = System.getProperties().stringPropertyNames().stream()
149                                  .filter(k -> k.startsWith("user."))
150                                  .collect(Collectors.toList());
151        keys.stream()
152            .map(k -> SystemPropertyName.newBuilder().setPropertyName(k).build())
153            .forEach(requestObserver::onNext);
154        requestObserver.onCompleted();
155        verify(responseObserver, timeout(100)).onNext(systemPropertiesCaptor.capture());
156
157        SystemProperties systemProperties = systemPropertiesCaptor.getValue();
158        systemProperties.getPropertiesMap()
159            .forEach((propertyName, propertyValue) ->
160            assertEquals(System.getProperty(propertyName), propertyValue));
161        verify(responseObserver, timeout(100)).onCompleted();
162        verify(responseObserver, never()).onError(any(Throwable.class));
163    }
164    // end::testGetClientStreamingProperties[]
165
166    @Test
167    // tag::testGetBidirectionalProperties[]
168    public void testGetBidirectionalProperties() {
169
170        int timesOnNext = 0;
171
172        @SuppressWarnings("unchecked")
173        StreamObserver<SystemProperty> responseObserver =
174            (StreamObserver<SystemProperty>) mock(StreamObserver.class);
175        StreamObserver<SystemPropertyName> requestObserver =
176            asyncStub.getBidirectionalProperties(responseObserver);
177
178        verify(responseObserver, never()).onNext(any(SystemProperty.class));
179
180        List<String> keys = System.getProperties().stringPropertyNames().stream()
181                                  .filter(k -> k.startsWith("java."))
182                                  .collect(Collectors.toList());
183
184        for (int i = 0; i < keys.size(); i++) {
185            SystemPropertyName spn = SystemPropertyName.newBuilder()
186                                         .setPropertyName(keys.get(i))
187                                         .build();
188            requestObserver.onNext(spn);
189            ArgumentCaptor<SystemProperty> systemPropertyCaptor =
190                ArgumentCaptor.forClass(SystemProperty.class);
191            verify(responseObserver, timeout(100).times(++timesOnNext))
192                .onNext(systemPropertyCaptor.capture());
193            SystemProperty systemProperty = systemPropertyCaptor.getValue();
194            assertEquals(System.getProperty(systemProperty.getPropertyName()),
195                         systemProperty.getPropertyValue());
196        }
197
198        requestObserver.onCompleted();
199        verify(responseObserver, timeout(100)).onCompleted();
200        verify(responseObserver, never()).onError(any(Throwable.class));
201    }
202    // end::testGetBidirectionalProperties[]
203}

  1// tag::copyright[]
  2/*******************************************************************************
  3 * Copyright (c) 2022, 2023 IBM Corporation and others.
  4 * All rights reserved. This program and the accompanying materials
  5 * are made available under the terms of the Eclipse Public License 2.0
  6 * which accompanies this distribution, and is available at
  7 * http://www.eclipse.org/legal/epl-2.0/
  8 *
  9 * SPDX-License-Identifier: EPL-2.0
 10 *******************************************************************************/
 11// end::copyright[]
 12package io.openliberty.guides.system;
 13
 14import java.util.HashMap;
 15import java.util.Map;
 16import java.util.logging.Logger;
 17
 18import io.grpc.stub.StreamObserver;
 19// tag::importGrpcClasses[]
 20import io.openliberty.guides.systemproto.SystemProperties;
 21import io.openliberty.guides.systemproto.SystemProperty;
 22import io.openliberty.guides.systemproto.SystemPropertyName;
 23import io.openliberty.guides.systemproto.SystemPropertyPrefix;
 24import io.openliberty.guides.systemproto.SystemPropertyValue;
 25import io.openliberty.guides.systemproto.SystemServiceGrpc;
 26// end::importGrpcClasses[]
 27
 28// tag::extends[]
 29public class SystemService extends SystemServiceGrpc.SystemServiceImplBase {
 30// end::extends[]
 31
 32    private static Logger logger = Logger.getLogger(SystemService.class.getName());
 33
 34    public SystemService() {
 35    }
 36
 37    // tag::getProperty[]
 38    @Override
 39    public void getProperty(
 40        SystemPropertyName request, StreamObserver<SystemPropertyValue> observer) {
 41
 42        // tag::pName[]
 43        String pName = request.getPropertyName();
 44        // end::pName[]
 45        // tag::pValue[]
 46        String pValue = System.getProperty(pName);
 47        // end::pValue[]
 48        // tag::response[]
 49        SystemPropertyValue value = SystemPropertyValue
 50                                        .newBuilder()
 51                                        .setPropertyValue(pValue)
 52                                        .build();
 53        // end::response[]
 54
 55        // tag::next[]
 56        observer.onNext(value);
 57        // end::next[]
 58        // tag::complete[]
 59        observer.onCompleted();
 60        // end::complete[]
 61
 62    }
 63    // end::getProperty[]
 64
 65    // tag::getServerStreamingProperties[]
 66    @Override
 67    public void getServerStreamingProperties(
 68        SystemPropertyPrefix request, StreamObserver<SystemProperty> observer) {
 69
 70        // tag::prefix[]
 71        String prefix = request.getPropertyPrefix();
 72        // end::prefix[]
 73        System.getProperties()
 74              .stringPropertyNames()
 75              .stream()
 76              // tag::filter[]
 77              .filter(name -> name.startsWith(prefix))
 78              // end::filter[]
 79              .forEach(name -> {
 80                  String pValue = System.getProperty(name);
 81                  // tag::serverMessage[]
 82                  SystemProperty value = SystemProperty
 83                      .newBuilder()
 84                      .setPropertyName(name)
 85                      .setPropertyValue(pValue)
 86                      .build();
 87                  // end::serverMessage[]
 88                  // tag::serverNext1[]
 89                  observer.onNext(value);
 90                  // end::serverNext1[]
 91                  logger.info("server streaming sent property: " + name);
 92               });
 93        // tag::serverComplete[]
 94        observer.onCompleted();
 95        // end::serverComplete[]
 96        logger.info("server streaming was completed!");
 97    }
 98    // end::getServerStreamingProperties[]
 99
100    // tag::getClientStreamingProperties[]
101    @Override
102    public StreamObserver<SystemPropertyName> getClientStreamingProperties(
103        StreamObserver<SystemProperties> observer) {
104
105        // tag::streamObserverClient[]
106        return new StreamObserver<SystemPropertyName>() {
107
108            // tag::clientStreamingMap[]
109            private Map<String, String> properties = new HashMap<String, String>();
110            // end::clientStreamingMap[]
111
112            // tag::receivingProperties[]
113            @Override
114            public void onNext(SystemPropertyName spn) {
115                String pName = spn.getPropertyName();
116                String pValue = System.getProperty(pName);
117                logger.info("client streaming received property: " + pName);
118                properties.put(pName, pValue);
119            }
120            // end::receivingProperties[]
121
122            @Override
123            public void onError(Throwable t) {
124                t.printStackTrace();
125            }
126
127            // tag::clientStreamingCompleted[]
128            @Override
129            public void onCompleted() {
130                SystemProperties value = SystemProperties.newBuilder()
131                                             .putAllProperties(properties)
132                                             .build();
133                observer.onNext(value);
134                observer.onCompleted();
135                logger.info("client streaming was completed!");
136            }
137            // end::clientStreamingCompleted[]
138        };
139        // end::streamObserverClient[]
140    }
141    // end::getClientStreamingProperties[]
142
143    // tag::getBidirectionalProperties[]
144    @Override
145    public StreamObserver<SystemPropertyName> getBidirectionalProperties(
146        StreamObserver<SystemProperty> observer) {
147
148        // tag::streamObserverBidirectional[]
149        return new StreamObserver<SystemPropertyName>() {
150            // tag::receiveBidirectionalProperties[]
151            @Override
152            public void onNext(SystemPropertyName spn) {
153                String pName = spn.getPropertyName();
154                String pValue = System.getProperty(pName);
155                logger.info("bi-directional streaming received: " + pName);
156                // tag::systemPropertyMessage[]
157                SystemProperty value = SystemProperty.newBuilder()
158                                           .setPropertyName(pName)
159                                           .setPropertyValue(pValue)
160                                           .build();
161                // end::systemPropertyMessage[]
162                // tag::serverNext2[]
163                observer.onNext(value);
164                // end::serverNext2[]
165            }
166            // end::receiveBidirectionalProperties[]
167
168            @Override
169            public void onError(Throwable t) {
170                t.printStackTrace();
171            }
172
173            // tag::bidirectionalCompleted[]
174            @Override
175            public void onCompleted() {
176                observer.onCompleted();
177                logger.info("bi-directional streaming was completed!");
178            }
179            // end::bidirectionalCompleted[]
180        };
181        // end::streamObserverBidirectional[]
182    }
183    // end::getBidirectionalProperties[]
184}

  1// tag::copyright[]
  2/*******************************************************************************
  3 * Copyright (c) 2022, 2023 IBM Corporation and others.
  4 * All rights reserved. This program and the accompanying materials
  5 * are made available under the terms of the Eclipse Public License 2.0
  6 * which accompanies this distribution, and is available at
  7 * http://www.eclipse.org/legal/epl-2.0/
  8 *
  9 * SPDX-License-Identifier: EPL-2.0
 10 *******************************************************************************/
 11// end::copyright[]
 12package it.io.openliberty.guides.query;
 13
 14import static org.junit.jupiter.api.Assertions.assertEquals;
 15import static org.junit.jupiter.api.Assertions.assertFalse;
 16
 17import java.net.MalformedURLException;
 18
 19import jakarta.json.JsonObject;
 20import jakarta.ws.rs.client.Client;
 21import jakarta.ws.rs.client.ClientBuilder;
 22import jakarta.ws.rs.client.WebTarget;
 23import jakarta.ws.rs.core.Response;
 24
 25import org.junit.jupiter.api.AfterAll;
 26import org.junit.jupiter.api.BeforeAll;
 27import org.junit.jupiter.api.Test;
 28
 29public class QueryIT {
 30
 31    private static final String PORT = System.getProperty("http.port", "9081");
 32    private static final String URL = "http://localhost:" + PORT + "/";
 33    private static Client client;
 34
 35    @BeforeAll
 36    public static void setup() {
 37        client = ClientBuilder.newClient();
 38    }
 39
 40    @AfterAll
 41    public static void teardown() {
 42        client.close();
 43    }
 44
 45    @Test
 46    // tag::getPropertiesString[]
 47    public void testGetPropertiesString() throws MalformedURLException {
 48        WebTarget target = client.target(URL + "query/properties/os.name");
 49        Response response = target.request().get();
 50        assertEquals(200, response.getStatus(),
 51                     "Incorrect response code from " + target.getUri().getPath());
 52        assertFalse(response.readEntity(String.class).isEmpty(),
 53                    "response should not be empty.");
 54        response.close();
 55    }
 56    // end::getPropertiesString[]
 57
 58    @Test
 59    // tag::getOSProperties[]
 60    public void testGetOSProperties() throws MalformedURLException {
 61        WebTarget target = client.target(URL + "query/properties/os");
 62        Response response = target.request().get();
 63        assertEquals(200, response.getStatus(),
 64                     "Incorrect response code from " + target.getUri().getPath());
 65        JsonObject obj = response.readEntity(JsonObject.class);
 66        assertFalse(obj.getString("os.name").isEmpty(),
 67                    "os.name should not be empty.");
 68        response.close();
 69    }
 70    // end::getOSProperties[]
 71
 72    @Test
 73    // tag::getUserProperties[]
 74    public void testGetUserProperties() throws MalformedURLException {
 75        WebTarget target = client.target(URL + "query/properties/user");
 76        Response response = target.request().get();
 77        assertEquals(200, response.getStatus(),
 78                     "Incorrect response code from " + target.getUri().getPath());
 79        JsonObject obj = response.readEntity(JsonObject.class);
 80        assertFalse(obj.getString("user.name").isEmpty(),
 81                    "user.name should not be empty.");
 82        response.close();
 83    }
 84    // end::getUserProperties[]
 85
 86    @Test
 87    // tag::getJavaProperties[]
 88    public void testGetJavaProperties() throws MalformedURLException {
 89        WebTarget target = client.target(URL + "query/properties/java");
 90        Response response = target.request().get();
 91        assertEquals(200, response.getStatus(),
 92                     "Incorrect response code from " + target.getUri().getPath());
 93        JsonObject obj = response.readEntity(JsonObject.class);
 94        assertFalse(obj.getString("java.home").isEmpty(),
 95                    "java.home should not be empty.");
 96        response.close();
 97    }
 98    // end::getJavaProperties[]
 99}