65 lines
2.2 KiB
Java
65 lines
2.2 KiB
Java
import io.aeron.Aeron;
|
|
import io.aeron.Subscription;
|
|
import io.aeron.Image;
|
|
import io.aeron.driver.MediaDriver;
|
|
import io.aeron.logbuffer.FragmentHandler;
|
|
import org.agrona.DirectBuffer;
|
|
import org.agrona.concurrent.BackoffIdleStrategy;
|
|
import org.agrona.concurrent.IdleStrategy;
|
|
|
|
import java.nio.charset.StandardCharsets;
|
|
import java.util.concurrent.atomic.AtomicBoolean;
|
|
|
|
public class Subscriber {
|
|
|
|
public static void main(String[] args) {
|
|
final AtomicBoolean running = new AtomicBoolean(true);
|
|
|
|
MediaDriver driver = MediaDriver.launchEmbedded();
|
|
|
|
Aeron.Context ctx = new Aeron.Context()
|
|
.availableImageHandler((Image img) -> System.out.println("Publisher available: " + img.sourceIdentity()))
|
|
.unavailableImageHandler((Image img) -> System.out.println("Publisher gone: " + img.sourceIdentity()))
|
|
.aeronDirectoryName(driver.aeronDirectoryName());
|
|
|
|
Aeron aeron = Aeron.connect(ctx);
|
|
|
|
String channel = "aeron:udp?endpoint=0.0.0.0:40123";
|
|
int streamId = 1001;
|
|
|
|
Subscription sub = aeron.addSubscription(channel, streamId);
|
|
IdleStrategy idleStrategy = new BackoffIdleStrategy(1, 1, 1, 1);
|
|
|
|
System.out.println("Subscriber ready, waiting for messages...");
|
|
|
|
// Fragment handler
|
|
FragmentHandler handler = (DirectBuffer buffer, int offset, int length, io.aeron.logbuffer.Header header) -> {
|
|
String msg = buffer.getStringWithoutLengthUtf8(offset, length).trim();
|
|
System.out.println("RAW MESSAGE: '" + msg + "'");
|
|
|
|
String[] parts = msg.split(",");
|
|
if (parts.length == 2) {
|
|
try {
|
|
long counter = Long.parseLong(parts[0].trim());
|
|
long sendTime = Long.parseLong(parts[1].trim());
|
|
long latency = System.currentTimeMillis() - sendTime;
|
|
|
|
System.out.printf("Received: %d | Latency: %d ms%n", counter, latency);
|
|
} catch (NumberFormatException e) {
|
|
System.err.println("Failed to parse message: " + msg);
|
|
}
|
|
} else {
|
|
System.err.println("Malformed message: " + msg);
|
|
}
|
|
};
|
|
|
|
while (running.get()) {
|
|
int fragments = sub.poll(handler, 10);
|
|
idleStrategy.idle(fragments);
|
|
}
|
|
|
|
sub.close();
|
|
aeron.close();
|
|
driver.close();
|
|
}
|
|
}
|