Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
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
2 changes: 1 addition & 1 deletion .github/workflows/build.yml
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ jobs:
DAPR_RUNTIME_VER: 1.6.0-rc.2
DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.6.0-rc.1/install/install.sh
DAPR_CLI_REF:
DAPR_REF: 5a307f3deaa1b322f7945179adad0403de80eb7e
DAPR_REF: 4cf499448ef6ee87c83db6a11b84e83237e92665
steps:
- uses: actions/checkout@v3
- name: Set up OpenJDK ${{ env.JDK_VER }}
Expand Down
2 changes: 1 addition & 1 deletion .github/workflows/validate.yml
Original file line number Diff line number Diff line change
Expand Up @@ -40,7 +40,7 @@ jobs:
DAPR_RUNTIME_VER: 1.6.0-rc.2
DAPR_INSTALL_URL: https://raw.githubusercontent.com/dapr/cli/v1.6.0-rc.1/install/install.sh
DAPR_CLI_REF:
DAPR_REF: 5a307f3deaa1b322f7945179adad0403de80eb7e
DAPR_REF: 4cf499448ef6ee87c83db6a11b84e83237e92665
steps:
- uses: actions/checkout@v3
- name: Set up OpenJDK ${{ env.JDK_VER }}
Expand Down
102 changes: 102 additions & 0 deletions sdk-tests/src/test/java/io/dapr/it/pubsub/http/PubSubIT.java
Original file line number Diff line number Diff line change
Expand Up @@ -37,20 +37,27 @@
import java.util.Collections;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Random;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Objects;

import static io.dapr.it.Retry.callWithRetry;
import static io.dapr.it.TestUtils.assertThrowsDaprException;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.assertNotNull;


@RunWith(Parameterized.class)
public class PubSubIT extends BaseIT {

private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

private static final TypeRef<List<CloudEvent>> CLOUD_EVENT_LIST_TYPE_REF = new TypeRef<>() {};
private static final TypeRef<List<CloudEvent<ConvertToLong>>> CLOUD_EVENT_LONG_LIST_TYPE_REF = new TypeRef<>() {};
private static final TypeRef<List<CloudEvent<MyObject>>> CLOUD_EVENT_MYOBJECT_LIST_TYPE_REF = new TypeRef<>() {};

//Number of messages to be sent: 10
Expand All @@ -66,6 +73,8 @@ public class PubSubIT extends BaseIT {
// Topic to test binary data
private static final String BINARY_TOPIC_NAME = "binarytopic";

private static final String LONG_TOPIC_NAME = "testinglongvalues";

/**
* Parameters for this test.
* Param #1: useGrpc.
Expand Down Expand Up @@ -402,6 +411,73 @@ public void testPubSubTTLMetadata() throws Exception {
daprRun.stop();
}

@Test
public void testLongValues() throws Exception {
final DaprRun daprRun = closeLater(startDaprApp(
this.getClass().getSimpleName(),
SubscriberService.SUCCESS_MESSAGE,
SubscriberService.class,
true,
60000));
// At this point, it is guaranteed that the service above is running and all ports being listened to.
if (this.useGrpc) {
daprRun.switchToGRPC();
} else {
daprRun.switchToHTTP();
}

ConvertToLong toLong = new ConvertToLong();
HashSet<ConvertToLong> expected = new HashSet<>();
Random random = new Random();
Long randomLong = 590518626939830271L;
random.setSeed(randomLong);
toLong.setValue(randomLong);
expected.add(toLong);
for (int i = 1; i < NUM_MESSAGES; i++) {
ConvertToLong value = new ConvertToLong();
randomLong = random.nextLong();
value.setValue(randomLong);
expected.add(value);
System.out.println("expected value is : " +value);
}
Iterator expectVal = expected.iterator();
try (DaprClient client = new DaprClientBuilder().build()) {
while(expectVal.hasNext()) {
//Publishing messages
client.publishEvent(
PUBSUB_NAME,
LONG_TOPIC_NAME,
expectVal.next(),
Collections.singletonMap(Metadata.TTL_IN_SECONDS, "1")).block();

try {
Thread.sleep((long) (1000 * Math.random()));
} catch (InterruptedException e) {
e.printStackTrace();
Thread.currentThread().interrupt();
return;
}
}
}

HashSet<ConvertToLong> actual = new HashSet<>();
try (DaprClient client = new DaprClientBuilder().build()) {
Comment thread
mukundansundar marked this conversation as resolved.
callWithRetry(() -> {
System.out.println("Checking results for topic " + LONG_TOPIC_NAME);
final List<CloudEvent<ConvertToLong>> messages = client.invokeMethod(
daprRun.getAppName(),
"messages/testinglongvalues",
null,
HttpExtension.GET, CLOUD_EVENT_LONG_LIST_TYPE_REF).block();
assertNotNull(messages);
for (int i = 0; i < NUM_MESSAGES; i++) {

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
for (int i = 0; i < NUM_MESSAGES; i++) {
Assert.assertNotNull(messages);
for (int i = 0; i < messages.size(); i++) {

actual.add(messages.get(i).getData());
}
assertEquals(expected,actual);
}, 2000);
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add assert comparing two set values

Assert.assertEquals(expected, actual);

Both expected and actual are of type Set<ConvertToLong

}

public static class MyObject {
private String id;

Expand All @@ -413,4 +489,30 @@ public void setId(String id) {
this.id = id;
}
}

public static class ConvertToLong {
private Long value;

public Long getValue() {
return value;
}

public void setValue(Long value) {
this.value = value;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ConvertToLong that = (ConvertToLong) o;
return Objects.equals(value, that.value);
}

@Override
public int hashCode() {
return Objects.hash(value);
}
}

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Add getter for value, and equals and Dashcode method (Can autogenerate in IntelliJ using Objects.hash etc ).


}
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,20 @@ public Mono<Void> handleMessageTTLTopic(@RequestBody(required = false) CloudEven
});
}

@Topic(name = "testinglongvalues", pubsubName = "messagebus")
@PostMapping(path = "/testinglongvalues")
public Mono<Void> handleMessageLongValues(@RequestBody(required = false) CloudEvent<PubSubIT.ConvertToLong> cloudEvent) {
return Mono.fromRunnable(() -> {
try {
Long message = cloudEvent.getData().getValue();
System.out.println("Subscriber got: " + message);
messagesByTopic.compute("testinglongvalues", merge(cloudEvent));
} catch (Exception e) {
throw new RuntimeException(e);
}
});
}

private BiFunction<String, List<CloudEvent<?>>, List<CloudEvent<?>>> merge(final CloudEvent<?> item) {
return (key, value) -> {
final List<CloudEvent<?>> list = value == null ? new ArrayList<>() : value;
Expand Down