Integrating RESTful services with a reactive system

  1// tag::copyright[]
  2/*******************************************************************************
  3 * Copyright (c) 2020, 2024 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.inventory;
 13
 14import java.util.List;
 15import java.util.Optional;
 16import java.util.Properties;
 17import java.util.logging.Logger;
 18import java.util.stream.Collectors;
 19
 20import jakarta.enterprise.context.ApplicationScoped;
 21import jakarta.inject.Inject;
 22import jakarta.ws.rs.Consumes;
 23import jakarta.ws.rs.DELETE;
 24import jakarta.ws.rs.GET;
 25import jakarta.ws.rs.PUT;
 26import jakarta.ws.rs.Path;
 27import jakarta.ws.rs.PathParam;
 28import jakarta.ws.rs.Produces;
 29import jakarta.ws.rs.core.MediaType;
 30import jakarta.ws.rs.core.Response;
 31
 32import org.eclipse.microprofile.reactive.messaging.Incoming;
 33import org.eclipse.microprofile.reactive.messaging.Outgoing;
 34import org.reactivestreams.Publisher;
 35
 36import io.openliberty.guides.models.PropertyMessage;
 37import io.openliberty.guides.models.SystemLoad;
 38import io.reactivex.rxjava3.core.BackpressureStrategy;
 39import io.reactivex.rxjava3.core.Flowable;
 40import io.reactivex.rxjava3.core.FlowableEmitter;
 41
 42
 43@ApplicationScoped
 44//tag::inventoryEndPoint[]
 45@Path("/inventory")
 46//end::inventoryEndPoint[]
 47public class InventoryResource {
 48
 49    private static Logger logger = Logger.getLogger(InventoryResource.class.getName());
 50    // tag::flowableEmitterDecl[]
 51    private FlowableEmitter<String> propertyNameEmitter;
 52    // end::flowableEmitterDecl[]
 53
 54    @Inject
 55    private InventoryManager manager;
 56
 57    @GET
 58    @Path("/systems")
 59    @Produces(MediaType.APPLICATION_JSON)
 60    public Response getSystems() {
 61        List<Properties> systems = manager.getSystems()
 62                                          .values()
 63                                          .stream()
 64                                          .collect(Collectors.toList());
 65        return Response.status(Response.Status.OK)
 66                       .entity(systems)
 67                       .build();
 68    }
 69
 70    @GET
 71    @Path("/systems/{hostname}")
 72    @Produces(MediaType.APPLICATION_JSON)
 73    public Response getSystem(@PathParam("hostname") String hostname) {
 74        Optional<Properties> system = manager.getSystem(hostname);
 75        if (system.isPresent()) {
 76            return Response.status(Response.Status.OK)
 77                           .entity(system)
 78                           .build();
 79        }
 80        return Response.status(Response.Status.NOT_FOUND)
 81                       .entity("hostname does not exist.")
 82                       .build();
 83    }
 84
 85    // tag::updateSystemProperty[]
 86    // tag::annotatedPut[]
 87    @PUT
 88    // end::annotatedPut[]
 89    // tag::putPath[]
 90    @Path("/data")
 91    // end::putPath[]
 92    @Produces(MediaType.APPLICATION_JSON)
 93    @Consumes(MediaType.TEXT_PLAIN)
 94    public Response updateSystemProperty(String propertyName) {
 95        logger.info("updateSystemProperty: " + propertyName);
 96        // tag::flowableEmitter[]
 97        propertyNameEmitter.onNext(propertyName);
 98        // end::flowableEmitter[]
 99        return Response
100                 .status(Response.Status.OK)
101                 .entity("Request successful for the " + propertyName + " property\n")
102                 .build();
103    }
104    // end::updateSystemProperty[]
105
106    @DELETE
107    @Produces(MediaType.APPLICATION_JSON)
108    public Response resetSystems() {
109        manager.resetSystems();
110        return Response.status(Response.Status.OK)
111                       .build();
112    }
113
114    // tag::updateStatus[]
115    // tag::systemLoad[]
116    @Incoming("systemLoad")
117    // end::systemLoad[]
118    public void updateStatus(SystemLoad sl)  {
119        String hostname = sl.hostname;
120        if (manager.getSystem(hostname).isPresent()) {
121            manager.updateCpuStatus(hostname, sl.loadAverage);
122            logger.info("Host " + hostname + " was updated: " + sl);
123        } else {
124            manager.addSystem(hostname, sl.loadAverage);
125            logger.info("Host " + hostname + " was added: " + sl);
126        }
127    }
128    // end::updateStatus[]
129
130    // tag::propertyMessage[]
131    @Incoming("addSystemProperty")
132    // end::propertyMessage[]
133    public void getPropertyMessage(PropertyMessage pm)  {
134        logger.info("getPropertyMessage: " + pm);
135        String hostId = pm.hostname;
136        if (manager.getSystem(hostId).isPresent()) {
137            manager.updatePropertyMessage(hostId, pm.key, pm.value);
138            logger.info("Host " + hostId + " was updated: " + pm);
139        } else {
140            manager.addSystem(hostId, pm.key, pm.value);
141            logger.info("Host " + hostId + " was added: " + pm);
142        }
143    }
144
145    // tag::sendPropertyName[]
146    // tag::OutgoingPropertyName[]
147    @Outgoing("requestSystemProperty")
148    // end::OutgoingPropertyName[]
149    public Publisher<String> sendPropertyName() {
150        // tag::flowableCreate[]
151        Flowable<String> flowable = Flowable.<String>create(emitter ->
152            this.propertyNameEmitter = emitter, BackpressureStrategy.BUFFER);
153        // end::flowableCreate[]
154        return flowable;
155    }
156    // end::sendPropertyName[]
157}

 1// tag::copyright[]
 2/*******************************************************************************
 3 * Copyright (c) 2020, 2024 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.lang.management.ManagementFactory;
15import java.lang.management.OperatingSystemMXBean;
16import java.net.InetAddress;
17import java.net.UnknownHostException;
18import java.util.concurrent.TimeUnit;
19import java.util.logging.Logger;
20
21import jakarta.enterprise.context.ApplicationScoped;
22
23import org.eclipse.microprofile.reactive.messaging.Incoming;
24import org.eclipse.microprofile.reactive.messaging.Outgoing;
25import org.reactivestreams.Publisher;
26
27import io.openliberty.guides.models.PropertyMessage;
28import io.openliberty.guides.models.SystemLoad;
29import io.reactivex.rxjava3.core.Flowable;
30
31@ApplicationScoped
32public class SystemService {
33
34    private static Logger logger = Logger.getLogger(SystemService.class.getName());
35
36    private static final OperatingSystemMXBean OS_MEAN =
37            ManagementFactory.getOperatingSystemMXBean();
38    private static String hostname = null;
39
40    private static String getHostname() {
41        if (hostname == null) {
42            try {
43                return InetAddress.getLocalHost().getHostName();
44            } catch (UnknownHostException e) {
45                return System.getenv("HOSTNAME");
46            }
47        }
48        return hostname;
49    }
50
51    // tag::publishSystemLoad[]
52    @Outgoing("systemLoad")
53    // end::publishSystemLoad[]
54    // tag::sendSystemLoad[]
55    public Publisher<SystemLoad> sendSystemLoad() {
56        // tag::flowableInterval[]
57        return Flowable.interval(15, TimeUnit.SECONDS)
58                       .map((interval -> new SystemLoad(getHostname(),
59                             OS_MEAN.getSystemLoadAverage())));
60        // end::flowableInterval[]
61    }
62    // end::sendSystemLoad[]
63
64    // tag::propertyRequest[]
65    @Incoming("propertyRequest")
66    // end::propertyRequest[]
67    // tag::propertyResponse[]
68    @Outgoing("propertyResponse")
69    // end::propertyResponse[]
70    // tag::sendProperty[]
71    public PropertyMessage sendProperty(String propertyName) {
72        logger.info("sendProperty: " + propertyName);
73        if (propertyName == null || propertyName.isEmpty()) {
74            logger.warning(propertyName == null ? "Null" : "An empty string"
75                + " is not System property.");
76            return null;
77        }
78        return new PropertyMessage(getHostname(),
79                       propertyName,
80                       System.getProperty(propertyName, "unknown"));
81    }
82    // end::sendProperty[]
83}