feat: add metrics bindings

roc_jni/src/main/impl/package.h

@@ -4,6 +4,7 @@
 
 #define CHANNEL_LAYOUT_CLASS PACKAGE_NAME "/ChannelLayout"
 #define CLOCK_SOURCE_CLASS PACKAGE_NAME "/ClockSource"
+#define CONNECTION_METRICS_CLASS PACKAGE_NAME "/RocConnectionMetrics"
 #define LATENCY_TUNER_BACKEND_CLASS PACKAGE_NAME "/LatencyTunerBackend"
 #define LATENCY_TUNER_PROFILE_CLASS PACKAGE_NAME "/LatencyTunerProfile"
 #define CONTEXT_CONFIG_CLASS PACKAGE_NAME "/RocContextConfig"
@@ -16,6 +17,8 @@
 #define PACKET_ENCODING_CLASS PACKAGE_NAME "/PacketEncoding"
 #define PROTOCOL_CLASS PACKAGE_NAME "/Protocol"
 #define RECEIVER_CONFIG_CLASS PACKAGE_NAME "/RocReceiverConfig"
+#define RECEIVER_METRICS_CLASS PACKAGE_NAME "/RocReceiverMetrics"
 #define RESAMPLER_BACKEND_CLASS PACKAGE_NAME "/ResamplerBackend"
 #define RESAMPLER_PROFILE_CLASS PACKAGE_NAME "/ResamplerProfile"
 #define SENDER_CONFIG_CLASS PACKAGE_NAME "/RocSenderConfig"
+#define SENDER_METRICS_CLASS PACKAGE_NAME "/RocSenderMetrics"

roc_jni/src/main/impl/receiver.c

@@ -4,6 +4,7 @@
 #include "exceptions.h"
 #include "helpers.h"
 #include "interface_config.h"
+#include "package.h"
 #include "receiver_config.h"
 
 #include <roc/receiver.h>
@@ -203,3 +204,89 @@ JNIEXPORT void JNICALL Java_org_rocstreaming_roctoolkit_RocReceiver_nativeReadFl
         (*env)->ReleaseFloatArrayElements(env, jsamples, samples, 0);
     }
 }
+
+JNIEXPORT jobject JNICALL Java_org_rocstreaming_roctoolkit_RocReceiver_nativeQuery(
+    JNIEnv* env, jobject jobj, jlong jreceiver, jint jslot) {
+    assert(env);
+
+    roc_receiver* receiver = (roc_receiver*) jreceiver;
+    roc_receiver_metrics receiver_metrics = {};
+    roc_connection_metrics* conn_metrics = NULL;
+    size_t conn_count = 0;
+    jobject jresult = NULL;
+    jclass conn_metrics_class = NULL;
+    jmethodID conn_metrics_constructor = NULL;
+    jobjectArray jconn_metrics_array = NULL;
+    jclass receiver_metrics_class = NULL;
+    jmethodID receiver_metrics_constructor = NULL;
+
+    if (!jreceiver) {
+        throw_exception(env, ILLEGAL_ARGUMENT_EXCEPTION, "Invalid RocReceiver: must not be null");
+        goto out;
+    }
+
+    if (roc_receiver_query(receiver, (roc_slot) jslot, &receiver_metrics, NULL, NULL) != 0) {
+        throw_exception(env, ROC_EXCEPTION, "Failed to query RocReceiver metrics");
+        goto out;
+    }
+
+    conn_count = receiver_metrics.connection_count;
+    if (conn_count > 0) {
+        conn_metrics = (roc_connection_metrics*) calloc(conn_count, sizeof(roc_connection_metrics));
+        if (!conn_metrics) {
+            throw_exception(env, ASSERTION_ERROR, "Failed to allocate memory for connection metrics");
+            goto out;
+        }
+
+        if (roc_receiver_query(receiver, (roc_slot) jslot, &receiver_metrics, conn_metrics, &conn_count) != 0) {
+            throw_exception(env, ROC_EXCEPTION, "Failed to query RocReceiver connection metrics");
+            goto out;
+        }
+    }
+
+    conn_metrics_class = find_class(env, CONNECTION_METRICS_CLASS);
+    if (!conn_metrics_class) {
+        goto out;
+    }
+
+    conn_metrics_constructor = find_method(
+        env, conn_metrics_class, "RocConnectionMetrics", "<init>", "(J)V");
+    if (!conn_metrics_constructor) {
+        goto out;
+    }
+
+    jconn_metrics_array = (*env)->NewObjectArray(env, (jsize) conn_count, conn_metrics_class, NULL);
+    if (!jconn_metrics_array) {
+        throw_exception(env, ASSERTION_ERROR, "Failed to create RocConnectionMetrics array");
+        goto out;
+    }
+
+    for (size_t i = 0; i < conn_count; i++) {
+        jobject jconn_metric = (*env)->NewObject(
+            env, conn_metrics_class, conn_metrics_constructor, (jlong) conn_metrics[i].e2e_latency);
+        if (!jconn_metric) {
+            throw_exception(env, ASSERTION_ERROR, "Failed to create RocConnectionMetrics object");
+            goto out;
+        }
+        (*env)->SetObjectArrayElement(env, jconn_metrics_array, (jsize) i, jconn_metric);
+        (*env)->DeleteLocalRef(env, jconn_metric);
+    }
+
+    receiver_metrics_class = find_class(env, RECEIVER_METRICS_CLASS);
+    if (!receiver_metrics_class) {
+        goto out;
+    }
+
+    receiver_metrics_constructor = find_method(
+        env, receiver_metrics_class, "RocReceiverMetrics", "<init>", "(I[L" CONNECTION_METRICS_CLASS ";)V");
+    if (!receiver_metrics_constructor) {
+        goto out;
+    }
+
+    jresult = (*env)->NewObject(env, receiver_metrics_class, receiver_metrics_constructor,
+        (jint) receiver_metrics.connection_count, jconn_metrics_array);
+
+out:
+    free(conn_metrics);
+    return jresult;
+}

roc_jni/src/main/impl/sender.c

@@ -4,6 +4,7 @@
 #include "exceptions.h"
 #include "helpers.h"
 #include "interface_config.h"
+#include "package.h"
 #include "sender_config.h"
 
 #include <roc/sender.h>
@@ -192,3 +193,89 @@ JNIEXPORT void JNICALL Java_org_rocstreaming_roctoolkit_RocSender_nativeWriteFlo
         (*env)->ReleaseFloatArrayElements(env, jsamples, samples, 0);
     }
 }
+
+JNIEXPORT jobject JNICALL Java_org_rocstreaming_roctoolkit_RocSender_nativeQuery(
+    JNIEnv* env, jobject jobj, jlong jsender, jint jslot) {
+    assert(env);
+
+    roc_sender* sender = (roc_sender*) jsender;
+    roc_sender_metrics sender_metrics = {};
+    roc_connection_metrics* conn_metrics = NULL;
+    size_t conn_count = 0;
+    jobject jresult = NULL;
+    jclass conn_metrics_class = NULL;
+    jmethodID conn_metrics_constructor = NULL;
+    jobjectArray jconn_metrics_array = NULL;
+    jclass sender_metrics_class = NULL;
+    jmethodID sender_metrics_constructor = NULL;
+
+    if (!jsender) {
+        throw_exception(env, ILLEGAL_ARGUMENT_EXCEPTION, "Invalid RocSender: must not be null");
+        goto out;
+    }
+
+    if (roc_sender_query(sender, (roc_slot) jslot, &sender_metrics, NULL, NULL) != 0) {
+        throw_exception(env, ROC_EXCEPTION, "Failed to query RocSender metrics");
+        goto out;
+    }
+
+    conn_count = sender_metrics.connection_count;
+    if (conn_count > 0) {
+        conn_metrics = (roc_connection_metrics*) calloc(conn_count, sizeof(roc_connection_metrics));
+        if (!conn_metrics) {
+            throw_exception(env, ASSERTION_ERROR, "Failed to allocate memory for connection metrics");
+            goto out;
+        }
+
+        if (roc_sender_query(sender, (roc_slot) jslot, &sender_metrics, conn_metrics, &conn_count) != 0) {
+            throw_exception(env, ROC_EXCEPTION, "Failed to query RocSender connection metrics");
+            goto out;
+        }
+    }
+
+    conn_metrics_class = find_class(env, CONNECTION_METRICS_CLASS);
+    if (!conn_metrics_class) {
+        goto out;
+    }
+
+    conn_metrics_constructor = find_method(
+        env, conn_metrics_class, "RocConnectionMetrics", "<init>", "(J)V");
+    if (!conn_metrics_constructor) {
+        goto out;
+    }
+
+    jconn_metrics_array = (*env)->NewObjectArray(env, (jsize) conn_count, conn_metrics_class, NULL);
+    if (!jconn_metrics_array) {
+        throw_exception(env, ASSERTION_ERROR, "Failed to create RocConnectionMetrics array");
+        goto out;
+    }
+
+    for (size_t i = 0; i < conn_count; i++) {
+        jobject jconn_metric = (*env)->NewObject(
+            env, conn_metrics_class, conn_metrics_constructor, (jlong) conn_metrics[i].e2e_latency);
+        if (!jconn_metric) {
+            throw_exception(env, ASSERTION_ERROR, "Failed to create RocConnectionMetrics object");
+            goto out;
+        }
+        (*env)->SetObjectArrayElement(env, jconn_metrics_array, (jsize) i, jconn_metric);
+        (*env)->DeleteLocalRef(env, jconn_metric);
+    }
+
+    sender_metrics_class = find_class(env, SENDER_METRICS_CLASS);
+    if (!sender_metrics_class) {
+        goto out;
+    }
+
+    sender_metrics_constructor = find_method(
+        env, sender_metrics_class, "RocSenderMetrics", "<init>", "(I[L" CONNECTION_METRICS_CLASS ";)V");
+    if (!sender_metrics_constructor) {
+        goto out;
+    }
+
+    jresult = (*env)->NewObject(env, sender_metrics_class, sender_metrics_constructor,
+        (jint) sender_metrics.connection_count, jconn_metrics_array);
+
+out:
+    free(conn_metrics);
+    return jresult;
+}

src/main/java/org/rocstreaming/roctoolkit/RocConnectionMetrics.java

@@ -0,0 +1,29 @@
+package org.rocstreaming.roctoolkit;
+
+import java.time.Duration;
+import lombok.*;
+
+/**
+ * Connection metrics.
+ * <p>
+ * Holds metrics for a single connection between sender and receiver.
+ */
+@Getter
+@Builder(builderClassName = "Builder", toBuilder = true)
+@ToString
+@EqualsAndHashCode
+public class RocConnectionMetrics {
+
+    /**
+     * Estimated end-to-end latency.
+     */
+    private final Duration e2eLatency;
+
+    public RocConnectionMetrics(Duration e2eLatency) {
+        this.e2eLatency = e2eLatency;
+    }
+
+    RocConnectionMetrics(long e2eLatencyNanos) {
+        this(Duration.ofNanos(e2eLatencyNanos));
+    }
+}

src/main/java/org/rocstreaming/roctoolkit/RocReceiver.java

@@ -385,6 +385,34 @@ public void read(float[] samples) throws RocException {
         nativeReadFloats(getPtr(), samples);
     }
 
+    /**
+     * Query receiver slot metrics.
+     *
+     * @param slot   specifies the receiver slot to query.
+     * @return       receiver metrics for the specified slot.
+     *
+     * @throws IllegalArgumentException   if the arguments are invalid.
+     * @throws RocException               if the slot does not exist or operation failed.
+     */
+    public RocReceiverMetrics query(Slot slot) throws RocException {
+        Check.notNull(slot, "Slot");
+
+        try {
+            LOGGER.log(Level.FINE, "entering RocReceiver.query(), ptr={0}, slot={1}",
+                    new Object[]{toHex(getPtr()), slot});
+
+            RocReceiverMetrics metrics = nativeQuery(getPtr(), slot.getValue());
+
+            LOGGER.log(Level.FINE, "leaving RocReceiver.query(), ptr={0}, metrics={1}",
+                    new Object[]{toHex(getPtr()), metrics});
+            return metrics;
+        } catch (Exception exc) {
+            LOGGER.log(Level.SEVERE, "exception in RocReceiver.query(), ptr={0}, exception={1}",
+                    new Object[]{toHex(getPtr()), exc});
+            throw exc;
+        }
+    }
+
     private static native long nativeOpen(long contextPtr, RocReceiverConfig config) throws RocException;
     private static native void nativeClose(long receiverPtr);
 
@@ -393,4 +421,5 @@ public void read(float[] samples) throws RocException {
     private native void nativeUnlink(long receiverPtr, int slot) throws RocException;
 
     private native void nativeReadFloats(long receiverPtr, float[] samples) throws RocException;
+    private native RocReceiverMetrics nativeQuery(long receiverPtr, int slot) throws RocException;
 }

src/main/java/org/rocstreaming/roctoolkit/RocReceiverMetrics.java

@@ -0,0 +1,33 @@
+package org.rocstreaming.roctoolkit;
+
+import lombok.*;
+
+/**
+ * Receiver metrics.
+ * <p>
+ * Holds receiver-side metrics that are not specific to connection.
+ * If multiple slots are used, each slot has its own metrics.
+ *
+ * @see RocReceiver
+ */
+@Getter
+@Builder(builderClassName = "Builder", toBuilder = true)
+@ToString
+@EqualsAndHashCode
+public class RocReceiverMetrics {
+
+    /**
+     * Number of active connections.
+     */
+    private final int connectionCount;
+
+    /**
+     * Metrics for individual active connections.
+     */
+    private final RocConnectionMetrics[] connectionMetrics;
+
+    public RocReceiverMetrics(int connectionCount, RocConnectionMetrics[] connectionMetrics) {
+        this.connectionCount = connectionCount;
+        this.connectionMetrics = connectionMetrics != null ? connectionMetrics : new RocConnectionMetrics[0];
+    }
+}

src/main/java/org/rocstreaming/roctoolkit/RocSender.java

@@ -358,6 +358,34 @@ public void write(float[] samples) throws RocException {
         nativeWriteFloats(getPtr(), samples);
     }
 
+    /**
+     * Query sender slot metrics.
+     *
+     * @param slot   specifies the sender slot to query.
+     * @return       sender metrics for the specified slot.
+     *
+     * @throws IllegalArgumentException   if the arguments are invalid.
+     * @throws RocException               if the slot does not exist or operation failed.
+     */
+    public RocSenderMetrics query(Slot slot) throws RocException {
+        Check.notNull(slot, "Slot");
+
+        try {
+            LOGGER.log(Level.FINE, "entering RocSender.query(), ptr={0}, slot={1}",
+                    new Object[]{toHex(getPtr()), slot});
+
+            RocSenderMetrics metrics = nativeQuery(getPtr(), slot.getValue());
+
+            LOGGER.log(Level.FINE, "leaving RocSender.query(), ptr={0}, metrics={1}",
+                    new Object[]{toHex(getPtr()), metrics});
+            return metrics;
+        } catch (Exception exc) {
+            LOGGER.log(Level.SEVERE, "exception in RocSender.query(), ptr={0}, exception={1}",
+                    new Object[]{toHex(getPtr()), exc});
+            throw exc;
+        }
+    }
+
     private static native long nativeOpen(long contextPtr, RocSenderConfig config) throws RocException;
     private static native void nativeClose(long senderPtr);
 
@@ -366,4 +394,5 @@ public void write(float[] samples) throws RocException {
     private native void nativeUnlink(long senderPtr, int slot) throws RocException;
 
     private native void nativeWriteFloats(long senderPtr, float[] samples) throws RocException;
+    private native RocSenderMetrics nativeQuery(long senderPtr, int slot) throws RocException;
 }