summaryrefslogtreecommitdiff
path: root/src/server
diff options
context:
space:
mode:
Diffstat (limited to 'src/server')
-rw-r--r--src/server/common.h32
-rw-r--r--src/server/server.c140
-rw-r--r--src/server/server.h2
3 files changed, 107 insertions, 67 deletions
diff --git a/src/server/common.h b/src/server/common.h
index 4d3a0ae..26d0edf 100644
--- a/src/server/common.h
+++ b/src/server/common.h
@@ -6,12 +6,12 @@
//#define CAMU_DB_PATH al_str_c("/home/andrew/c/camu/data/camu_db_test")
#define CAMU_PORT 14356
-#define CAMU_RESOURCE_PORT 14357
+#define CAMU_MULTIPLEX_RPC 0x53
+#define CAMU_MULTIPLEX_SHRUB 0x85
-#define CAMU_SERVER_IP al_str_c("127.0.0.1")
+//#define CAMU_SERVER_IP al_str_c("127.0.0.1")
//#define CAMU_SERVER_IP al_str_c("192.168.1.192")
-//#define CAMU_SERVER_IP al_str_c("108.52.160.112")
-
+#define CAMU_SERVER_IP al_str_c("108.52.160.112")
enum {
CAMU_NODE = 0,
@@ -20,23 +20,15 @@ enum {
};
enum {
- CAMU_SRV_IDENTIFY = 0,
- CAMU_SRV_LIST_ACTION,
- CAMU_SRV_CREATE_SEARCH,
- CAMU_SRV_GET_PAGE,
- CAMU_SRV_CREATE_BROWSE,
- CAMU_SRV_GET_PATH
-};
-
-enum {
- CAMU_CONN_STATE
+ CAMU_SERVER_IDENTIFY = 0,
+ CAMU_SERVER_LIST_ACTION
};
enum {
- CAMU_LIST_ADD = 0,
- CAMU_LIST_SKIP,
- CAMU_LIST_SKIPTO,
- CAMU_LIST_TOGGLE_PAUSE,
- CAMU_LIST_SEEK,
- CAMU_LIST_FINISHED
+ CAMU_ADD = 0,
+ CAMU_SKIP,
+ CAMU_SKIPTO,
+ CAMU_TOGGLE_PAUSE,
+ CAMU_SEEK,
+ CAMU_FINISHED
};
diff --git a/src/server/server.c b/src/server/server.c
index 0fd1d71..4e11d3e 100644
--- a/src/server/server.c
+++ b/src/server/server.c
@@ -1,7 +1,6 @@
#include <al/log.h>
#include "../libsink/common.h"
-#include "../shrub/common.h"
#include "server.h"
#include "common.h"
@@ -21,16 +20,23 @@ static struct camu_user *get_user_by_username(struct camu_server *tree, str *use
return NULL;
}
-static void list_callback(void *userdata, u8 op, void *opaque, s32 sequence)
+static void list_callback(void *userdata, u8 op, void *opaque, void *prev_opaque, s32 sequence)
{
struct camu_srv_sink *sink = (struct camu_srv_sink *)userdata;
struct camu_srv_resource *resource = (struct camu_srv_resource *)opaque;
+ struct camu_srv_resource *prev = (struct camu_srv_resource *)prev_opaque;
struct aki_packet *packet = aki_rpc_get_packet(sink->conn->rpc, op);
aki_packet_write_str(packet, CAMU_SERVER_IP);
- aki_packet_write_s32(packet, CAMU_RESOURCE_PORT);
+ aki_packet_write_s32(packet, CAMU_PORT);
aki_packet_write_u16(packet, resource->node->id);
aki_packet_write_s32(packet, sequence);
aki_packet_write_u64(packet, resource->node->start);
+ aki_packet_write_u8(packet, resource->node->paused != SHRUB_NOT_PAUSED);
+ if (prev) {
+ aki_packet_write_u64(packet, prev->node->paused_at);
+ } else {
+ aki_packet_write_u64(packet, 0);
+ }
aki_rpc_connection_command(sink->conn, packet, NULL, NULL);
if (op == CAMU_SINK_SET) {
al_printf("Now Playing: %.*s\n", AL_STR_PRINTF(&resource->unique_id));
@@ -81,7 +87,52 @@ static bool identify_command_callback(void *userdata, struct aki_rpc_connection
return true;
}
-static bool list_command_callback(void *userdata, struct aki_rpc_connection *conn,
+static void list_entry_callback(void *entry, u8 op, void *opaque)
+{
+ struct camu_srv_resource *resource = (struct camu_srv_resource *)entry;
+ switch (op) {
+ case CAMU_LIST_ENTRY_ID:
+ *(str **)opaque = &resource->unique_id;
+ break;
+ case CAMU_LIST_IMPULSE:
+ shrb_node_set_start(resource->node, *(u64 *)opaque);
+ break;
+ case CAMU_LIST_USER_PAUSE:
+ shrb_node_user_pause(resource->node, *(u64 *)opaque);
+ break;
+ case CAMU_LIST_USER_RESUME: {
+ struct shrb_resume_req *req = (struct shrb_resume_req *)opaque;
+ shrb_node_user_resume(resource->node, req);
+ break;
+ }
+ case CAMU_LIST_PAUSE:
+ shrb_node_pause(resource->node);
+ break;
+ case CAMU_LIST_SEEK:
+ shrb_node_seek(resource->node, *(u64 *)opaque);
+ break;
+ }
+}
+
+#ifdef CAMU_LOCAL_SOCKET
+static void list_add_local(struct camu_server *srv, struct camu_list *list, str *line)
+{
+ struct cch_entry *entry = entry_for_external_path(&srv->portal, line);
+ if (entry) {
+ struct shrb_node *node = shrb_server_create_node(&srv->resource, 0, entry);
+ if (node) {
+ struct camu_srv_resource *resource = al_alloc_object(struct camu_srv_resource);
+ al_array_push(srv->resources, resource);
+ al_str_clone(&resource->unique_id, line);
+ resource->entry = entry;
+ resource->node = node;
+ camu_list_add(list, resource, list_entry_callback, resource->node->duration, false);
+ }
+ }
+}
+#endif
+
+static bool list_action_command_callback(void *userdata, struct aki_rpc_connection *conn,
struct aki_packet *packet, struct aki_packet *rpacket)
{
struct camu_server *srv = (struct camu_server *)userdata;
@@ -92,33 +143,38 @@ static bool list_command_callback(void *userdata, struct aki_rpc_connection *con
struct camu_list *list = al_array_at(user->lists, 0);
switch (aki_packet_read_u8(packet)) {
- case CAMU_LIST_ADD: {
+ case CAMU_ADD: {
+ str line;
+ aki_packet_read_str(packet, &line);
+#ifdef CAMU_LOCAL_SOCKET
+ list_add_local(srv, list, &line);
+#endif
break;
}
- case CAMU_LIST_SKIP: {
+ case CAMU_SKIP: {
s32 sequence = aki_packet_read_s32(packet);
s32 n = aki_packet_read_s32(packet);
camu_list_skip(list, sequence, n);
break;
}
- case CAMU_LIST_SKIPTO: {
+ case CAMU_SKIPTO: {
s32 i = aki_packet_read_s32(packet);
camu_list_skipto(list, i);
break;
}
- case CAMU_LIST_TOGGLE_PAUSE: {
+ case CAMU_TOGGLE_PAUSE: {
s32 sequence = aki_packet_read_s32(packet);
u64 pos = aki_packet_read_u64(packet);
camu_list_toggle_pause(list, sequence, pos);
break;
}
- case CAMU_LIST_SEEK: {
+ case CAMU_SEEK: {
s32 sequence = aki_packet_read_s32(packet);
f64 percent = aki_packet_read_f64(packet);
camu_list_seek(list, sequence, percent);
break;
}
- case CAMU_LIST_FINISHED: {
+ case CAMU_FINISHED: {
s32 sequence = aki_packet_read_s32(packet);
camu_list_finished(list, sequence);
break;
@@ -130,10 +186,8 @@ static bool list_command_callback(void *userdata, struct aki_rpc_connection *con
}
static struct aki_rpc_command commands[] = {
- { .op = CAMU_SRV_IDENTIFY, .callback = identify_command_callback, .userdata = NULL },
- { .op = CAMU_SRV_LIST_ACTION, .callback = list_command_callback, .userdata = NULL },
- //{ .op = CAMU_SRV_CREATE_SEARCH, .callback = create_search_command_callback, .userdata = NULL },
- //{ .op = CAMU_SRV_GET_PAGE, .callback = get_page_command_callback, .userdata = NULL },
+ { .op = CAMU_SERVER_IDENTIFY, .callback = identify_command_callback, .userdata = NULL },
+ { .op = CAMU_SERVER_LIST_ACTION, .callback = list_action_command_callback, .userdata = NULL }
};
static void connection_callback(void *userdata, struct aki_rpc_connection *conn)
@@ -195,23 +249,6 @@ static void connection_closed_callback(void *userdata, struct aki_rpc_connection
}
}
-static void list_entry_callback(void *entry, u8 op, void *opaque)
-{
- struct camu_srv_resource *resource = (struct camu_srv_resource *)entry;
- switch (op) {
- case CAMU_LIST_ENTRY_IMPULSE: {
- struct camu_list_timing *timing = (struct camu_list_timing *)opaque;
- shrb_node_set_start(resource->node, timing->start);
- break;
- }
- case CAMU_LIST_ENTRY_TOGGLE_PAUSE:
- shrb_node_toggle_pause(resource->node, *(u64 *)opaque);
- break;
- case CAMU_LIST_ENTRY_SEEK:
- shrb_node_seek(resource->node, *(u64 *)opaque);
- break;
- }
-}
#ifdef CAMU_LOCAL_SOCKET
static u8 server_line_callback(void *userdata, str *line)
@@ -226,20 +263,12 @@ static u8 server_line_callback(void *userdata, str *line)
} else if (al_str_eq(line, al_str_c(";SHUFFLE"))) {
camu_list_shuffle(list);
} else if (al_str_eq(line, al_str_c(";SORT"))) {
+ camu_list_sort(list);
+ } else if (al_str_eq(line, al_str_c(";REVERSE"))) {
+ camu_list_reverse(list);
} else if (al_str_eq(line, al_str_c(";CLEAR"))) {
} else {
- struct cch_entry *entry = entry_for_external_path(&srv->portal, line);
- if (entry) {
- struct shrb_node *node = shrb_server_create_node(&srv->resource, 0, entry);
- if (node) {
- struct camu_srv_resource *resource = al_alloc_object(struct camu_srv_resource);
- al_array_push(srv->resources, resource);
- al_str_clone(&resource->unique_id, line);
- resource->entry = entry;
- resource->node = node;
- camu_list_add(list, resource, list_entry_callback, resource->node->duration, false);
- }
- }
+ list_add_local(srv, list, line);
}
return AKI_LINE_PROCESSOR_CONTINUE;
}
@@ -254,6 +283,22 @@ static void sigint_handler(s32 signum)
static struct camu_server srv = { 0 };
+static bool multiplex_callback(void *userdata, u8 id, struct aki_socket *s)
+{
+ struct camu_server *srv = (struct camu_server *)userdata;
+ switch (id) {
+ case CAMU_MULTIPLEX_RPC:
+ aki_rpc_add_socket(&srv->server, s);
+ return true;
+ case CAMU_MULTIPLEX_SHRUB:
+ shrb_server_add_socket(&srv->resource, s);
+ return true;
+ default:
+ break;
+ }
+ return false;
+}
+
s32 main(void)
{
aki_common_init();
@@ -275,15 +320,16 @@ s32 main(void)
aki_event_loop_init(&srv.loop);
- aki_rpc_init(&srv.server, AKI_SOCKET_TCP, connection_callback, connection_closed_callback, &srv);
+ aki_rpc_init(&srv.server, &srv.loop, connection_callback, connection_closed_callback, &srv);
for (u32 i = 0; i < AL_ARRAY_SIZE(commands); i++) {
commands[i].userdata = &srv;
aki_rpc_add_command(&srv.server, &commands[i]);
}
- aki_rpc_listen(&srv.server, &srv.loop, al_str_c("0.0.0.0"), CAMU_PORT);
- shrb_server_init(&srv.resource);
- shrb_server_listen(&srv.resource, &srv.loop, al_str_c("0.0.0.0"), CAMU_RESOURCE_PORT);
+ shrb_server_init(&srv.resource, &srv.loop);
+
+ aki_multiplex_socket_init(&srv.multi, AKI_SOCKET_TCP, multiplex_callback, &srv);
+ aki_multiplex_socket_listen(&srv.multi, &srv.loop, al_str_c("0.0.0.0"), CAMU_PORT);
#ifdef CAMU_LOCAL_SOCKET
srv.socket.type = AKI_SOCKET_UNIX;
diff --git a/src/server/server.h b/src/server/server.h
index 83e8523..9565bf6 100644
--- a/src/server/server.h
+++ b/src/server/server.h
@@ -2,6 +2,7 @@
#define CAMU_LOCAL_SOCKET
+#include <aki/multiplex.h>
#include <aki/rpc2.h>
#ifdef CAMU_LOCAL_SOCKET
#include <aki/line_processor.h>
@@ -38,6 +39,7 @@ struct camu_srv_resource {
struct camu_server {
struct aki_event_loop loop;
+ struct aki_multiplex_socket multi;
struct aki_rpc server;
array(struct camu_srv_node *) nodes;
array(struct camu_srv_client *) clients;
generated by cgit v1.2.3-101-g0448 (git 2.54.0) at 2026-07-25 20:07:30 +0000