Function runtime_worker_handle_subscribe

src/daemon/runtime.c:1200–1226

}

static bool runtime_worker_handle_subscribe(cbm_daemon_runtime_worker_t *worker,
                                            const uint8_t *payload, uint32_t length) {
    if (!runtime_worker_service_running(worker) || !payload ||
        length < SUBSCRIBE_REQUEST_PREFIX_SIZE) {
        return false;
    }
    uint32_t key_length = runtime_get_u32(payload);
    if (key_length == 0 || key_length > CBM_DAEMON_RUNTIME_PROJECT_KEY_MAX ||
        length != SUBSCRIBE_REQUEST_PREFIX_SIZE + key_length) {
        return false;
    }
    char project_key[CBM_DAEMON_RUNTIME_PROJECT_KEY_MAX + 1];
    memcpy(project_key, payload + SUBSCRIBE_REQUEST_PREFIX_SIZE, key_length);
    project_key[key_length] = '\0';
    if (memchr(project_key, '\0', key_length) != NULL) {
        return false;
    }
    cbm_daemon_subscription_id_t subscription_id = CBM_DAEMON_SUBSCRIPTION_ID_INVALID;
    cbm_daemon_subscription_result_t result = cbm_daemon_job_subscribe(
        worker->service->coordinator, worker->client_id, project_key, &subscription_id);
    uint8_t response[SUBSCRIBE_RESPONSE_SIZE];
    runtime_put_u32(response, (uint32_t)result);
    runtime_put_u64(response + 4, subscription_id);
    return runtime_worker_send_frame(worker, CBM_DAEMON_FRAME_RESPONSE,
                                     CBM_DAEMON_RUNTIME_OP_JOB_SUBSCRIBE, response,
                                     (uint32_t)sizeof(response));
}

static bool runtime_worker_handle_unsubscribe(cbm_daemon_runtime_worker_t *worker,
                                              const uint8_t *payload, uint32_t length) {