homatthew commented on code in PR #3539: URL: https://github.com/apache/gobblin/pull/3539#discussion_r953878946
########## gobblin-runtime/src/main/java/org/apache/gobblin/runtime/messaging/DynamicWorkUnitUtils.java: ########## @@ -0,0 +1,94 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.gobblin.runtime.messaging; + +import com.google.common.base.Optional; +import com.google.gson.Gson; +import com.google.gson.JsonElement; +import com.google.gson.JsonObject; +import java.time.Duration; +import java.time.Instant; +import java.time.temporal.ChronoUnit; +import lombok.NonNull; +import org.apache.gobblin.runtime.messaging.data.DynamicWorkUnitMessage; +import org.apache.gobblin.util.LoggingUncaughtExceptionHandler; + + +public final class DynamicWorkUnitUtils { + private static final Gson GSON = new Gson(); + private static final String PROPS_PREFIX = "DynamicWorkUnit.Props"; + private static final String MESSAGE_IMPLEMENTATION = PROPS_PREFIX + ".MessageImplementationClass"; + + private DynamicWorkUnitUtils() { + throw new RuntimeException("Cannot instantiate utils class"); + } + + /** + * Helper method for deserializing {@link DynamicWorkUnitMessage} + * @param json Serialized message using {@link DynamicWorkUnitUtils#toJson} + * @return {@link DynamicWorkUnitMessage} POJO representation of the given json + */ + public static <T extends DynamicWorkUnitMessage> DynamicWorkUnitMessage fromJson(JsonElement json) { + try { + JsonObject obj = json.getAsJsonObject(); + Class<T> clazz = (Class<T>) Class.forName(obj.get(MESSAGE_IMPLEMENTATION).getAsString()); + return GSON.fromJson(json, clazz); + } catch (IllegalStateException | ClassNotFoundException e) { + throw new RuntimeException(e); + } + } + + /** + * Helper method for serializing {@link DynamicWorkUnitMessage} + * @param msg Message object to serialize + * @return json representation of message object + */ + public static JsonElement toJson(@NonNull DynamicWorkUnitMessage msg) { + JsonElement json = GSON.toJsonTree(msg); + JsonObject obj = json.getAsJsonObject(); + obj.addProperty(MESSAGE_IMPLEMENTATION, msg.getClass().getName()); + return obj; + } + + /** + * Helper method for periodically executing runnable. Useful method to allow {@link DynamicWorkUnitConsumer} to poll + * for new messages + * @param runnable + * @param timeBetweenRunStarts + */ + public static void runInBackground(Runnable runnable, Duration timeBetweenRunStarts) { + Thread idleProcessThread = new Thread(() -> { + while (true) { + Instant nextStartTime = Instant.now().plus(timeBetweenRunStarts); + runnable.run(); + try { + long timeToSleepMs = Instant.now().until(nextStartTime, ChronoUnit.MILLIS); + if(timeToSleepMs > 0) { Review Comment: Fun fact. This https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ScheduledExecutorService.html class does what this method does but better. I will remove it in the next iteration. This method's behavior is the exact same as `scheduleAtFixedRate` method. No minimum wait time is because this method was intended to be used with a message consumer. If the time of execution exceeds the `timeBetweenRunStarts` time, then we probably have a backlog of messages that still need to be processed (hence we want to run the consumer again). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
