Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Include complete state update in udmi_target #758

Merged
merged 3 commits into from
Nov 8, 2023
Merged

Include complete state update in udmi_target #758

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
eddb8b6
Republish complete state update
grafnu Nov 7, 2023
d27700f
Fix error message
grafnu Nov 7, 2023
c8dd780
Fix linty
grafnu Nov 7, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions udmis/src/main/java/com/google/bos/udmi/service/core/StateProcessor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,7 @@ private void shardStateUpdate(MessageContinuation continuation, StateUpdate mess
envelope.subType = SubType.STATE;
envelope.subFolder = UPDATE;
reflectMessage(envelope, stringify(message));
continuation.publish(message);
String originalTransaction = envelope.transactionId;
AtomicInteger txnSuffix = new AtomicInteger();
info("Sharding state message for %s/%s %s", envelope.deviceRegistryId, envelope.deviceId,
Expand Down
39 changes: 22 additions & 17 deletions udmis/src/test/java/com/google/bos/udmi/service/core/StateProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,13 @@
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;

import com.google.bos.udmi.service.messaging.StateUpdate;
import com.google.bos.udmi.service.messaging.impl.MessageBase;
import com.google.bos.udmi.service.messaging.impl.MessageBase.Bundle;
import com.google.udmi.util.CleanDateFormat;
import java.util.Date;
import java.util.function.Function;
import java.util.function.Predicate;
import org.jetbrains.annotations.NotNull;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
Expand All @@ -44,6 +46,10 @@ protected Class<? extends ProcessorBase> getProcessorClass() {
return StateProcessor.class;
}

private boolean contains(Predicate<Object> objectPredicate) {
return captured.stream().anyMatch(objectPredicate);
}

private Config getTestConfig() {
Config config = new Config();
config.system = new SystemConfig();
Expand Down Expand Up @@ -123,6 +129,15 @@ public void lastStartUpdate() {
assertNull(newConfig, "new last_start");
}

@Test
public void legacyStateMessage() {
StateProcessor processor = initializeTestInstance(StateProcessor.class);
Object message = loadFileRequired(Object.class, LEGACY_STATE_MESSAGE_FILE);
dispatcher.withEnvelopeFor(new Envelope(), message, () -> processor.defaultHandler(message));
getReverseDispatcher().waitForMessageProcessed(SystemState.class);
terminateAndWait();
}

/**
* Test that a state update with multiple sub-blocks results in the expected two messages.
*/
Expand All @@ -134,24 +149,14 @@ public void multiExpansion() {
getReverseDispatcher().waitForMessageProcessed(GatewayState.class);
terminateAndWait();

assertEquals(2, captured.size(), "unexpected received message count");
assertTrue(captured.stream().anyMatch(message -> message instanceof SystemState),
"has SystemState");
assertTrue(captured.stream().anyMatch(message -> message instanceof GatewayState),
"has GatewayState");
assertEquals(3, captured.size(), "unexpected received message count");
assertTrue(contains(message -> message instanceof StateUpdate), "has StateUpdate");
assertTrue(contains(message -> message instanceof SystemState), "has SystemState");
assertTrue(contains(message -> message instanceof GatewayState), "has GatewayState");
assertEquals(0, getExceptionCount(), "exception count");
assertEquals(1, getDefaultCount(), "default handler count");
}

@Test
public void legacyStateMessage() {
StateProcessor processor = initializeTestInstance(StateProcessor.class);
Object message = loadFileRequired(Object.class, LEGACY_STATE_MESSAGE_FILE);
dispatcher.withEnvelopeFor(new Envelope(), message, () -> processor.defaultHandler(message));
getReverseDispatcher().waitForMessageProcessed(SystemState.class);
terminateAndWait();
}

/**
* Test that a state update with one sub-block results in a received message of the proper type.
*/
Expand All @@ -162,9 +167,9 @@ public void singleExpansion() {
safeSleep(ASYNC_PROCESSING_DELAY_MS);
terminateAndWait();

assertEquals(1, captured.size(), "unexpected received message count");
Object received = captured.get(0);
assertTrue(received instanceof SystemState, "expected SystemState message");
assertEquals(2, captured.size(), "unexpected received message count");
assertTrue(contains(msg -> msg instanceof SystemState), "expected SystemState message");
assertTrue(contains(msg -> msg instanceof StateUpdate), "expected StateUpdate message");
assertEquals(0, getExceptionCount(), "exception count");
assertEquals(1, getDefaultCount(), "default handler count");

Expand Down
Loading