Skip to content

Commit 3e879fe

Browse files
timkpainetimkpaine
committed
More wip
Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com>
1 parent cfea0d8 commit 3e879fe

33 files changed

Lines changed: 8874 additions & 71 deletions

‎cpp/csp/adapters/c/example/CMakeLists.txt‎

Lines changed: 23 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,20 +1,38 @@
1+
# Example C adapters demonstrating the C ABI interface
12

23
set(C_EXAMPLE_HEADER_FILES
34
ExamplePushInputAdapter.h
45
ExampleOutputAdapter.h
6+
ExampleManagedAdapter.h
57
)
68

79
set(C_EXAMPLE_SOURCE_FILES
810
ExamplePushInputAdapter.c
911
ExampleOutputAdapter.c
10-
${KAFKA_HEADER_FILES}
12+
ExampleManagedAdapter.c
1113
)
1214

13-
add_library(csp_c_example_adapter STATIC ${C_EXAMPLE_SOURCE_FILES})
14-
set_target_properties(csp_c_example_adapter PROPERTIES PUBLIC_HEADER "${C_EXAMPLE_HEADER_FILES}")
15+
add_library(csp_c_example_adapter STATIC ${C_EXAMPLE_SOURCE_FILES} ${C_EXAMPLE_HEADER_FILES})
16+
17+
# Include the engine c headers
18+
target_include_directories(csp_c_example_adapter PUBLIC
19+
${CMAKE_SOURCE_DIR}/cpp
20+
)
21+
22+
# Set C standard
23+
set_target_properties(csp_c_example_adapter PROPERTIES
24+
PUBLIC_HEADER "${C_EXAMPLE_HEADER_FILES}"
25+
C_STANDARD 11
26+
C_STANDARD_REQUIRED ON
27+
)
28+
29+
# On non-Windows, we need pthread for the example threaded adapters
30+
if(NOT WIN32)
31+
target_link_libraries(csp_c_example_adapter pthread)
32+
endif()
1533

1634
install(TARGETS csp_c_example_adapter
17-
PUBLIC_HEADER DESTINATION include/csp/adapters/example
35+
PUBLIC_HEADER DESTINATION include/csp/adapters/c/example
1836
RUNTIME DESTINATION ${CSP_RUNTIME_INSTALL_SUBDIR}
1937
LIBRARY DESTINATION lib/
20-
)
38+
)

‎cpp/csp/adapters/c/example/ExampleManagedAdapter.c‎

Lines changed: 207 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,207 @@
1+
/*
2+
* Example Managed Adapter Implementation
3+
*
4+
* Demonstrates how to build an adapter manager that coordinates
5+
* multiple input/output adapters, similar to KafkaAdapterManager.
6+
*/
7+
8+
#include "ExampleManagedAdapter.h"
9+
#include <stdio.h>
10+
#include <stdlib.h>
11+
#include <string.h>
12+
13+
/* ============================================================================
14+
* Adapter Manager Callbacks
15+
* ============================================================================ */
16+
17+
static const char* managed_adapter_name(void* user_data)
18+
{
19+
ManagedAdapterState* state = (ManagedAdapterState*)user_data;
20+
return state->name;
21+
}
22+
23+
static void managed_adapter_start(void* user_data, CCspAdapterManagerHandle manager,
24+
CCspDateTime start_time, CCspDateTime end_time)
25+
{
26+
ManagedAdapterState* state = (ManagedAdapterState*)user_data;
27+
state->manager = manager;
28+
state->is_started = 1;
29+
state->message_count = 0;
30+
31+
/* Calculate time in seconds for display */
32+
double start_sec = (double)start_time / 1000000000.0;
33+
double end_sec = (double)end_time / 1000000000.0;
34+
35+
printf("[%s] Manager started (start=%.3f, end=%.3f)\n",
36+
state->name, start_sec, end_sec);
37+
38+
/* Report status to the graph */
39+
ccsp_adapter_manager_push_status(manager, CCSP_STATUS_LEVEL_INFO, 0,
40+
"Manager started successfully");
41+
}
42+
43+
static void managed_adapter_stop(void* user_data)
44+
{
45+
ManagedAdapterState* state = (ManagedAdapterState*)user_data;
46+
47+
printf("[%s] Manager stopped. Total messages: %d\n",
48+
state->name, state->message_count);
49+
50+
state->is_started = 0;
51+
}
52+
53+
static CCspDateTime managed_adapter_process_next_sim_time_slice(void* user_data,
54+
CCspDateTime time)
55+
{
56+
/* This example is realtime-only, so we return 0 (no more sim data) */
57+
(void)user_data;
58+
(void)time;
59+
return 0;
60+
}
61+
62+
static void managed_adapter_destroy(void* user_data)
63+
{
64+
ManagedAdapterState* state = (ManagedAdapterState*)user_data;
65+
printf("[%s] Manager destroyed\n", state->name);
66+
free(state);
67+
}
68+
69+
CCspAdapterManagerVTable example_managed_adapter_create(const char* name)
70+
{
71+
ManagedAdapterState* state = (ManagedAdapterState*)malloc(sizeof(ManagedAdapterState));
72+
if (!state) {
73+
CCspAdapterManagerVTable empty = {0};
74+
return empty;
75+
}
76+
77+
memset(state, 0, sizeof(ManagedAdapterState));
78+
if (name) {
79+
strncpy(state->name, name, sizeof(state->name) - 1);
80+
} else {
81+
strncpy(state->name, "ExampleManagedAdapter", sizeof(state->name) - 1);
82+
}
83+
84+
CCspAdapterManagerVTable vtable;
85+
vtable.user_data = state;
86+
vtable.name = managed_adapter_name;
87+
vtable.start = managed_adapter_start;
88+
vtable.stop = managed_adapter_stop;
89+
vtable.process_next_sim_time_slice = managed_adapter_process_next_sim_time_slice;
90+
vtable.destroy = managed_adapter_destroy;
91+
92+
return vtable;
93+
}
94+
95+
/* ============================================================================
96+
* Managed Output Adapter Callbacks
97+
* ============================================================================ */
98+
99+
static void managed_output_start(void* user_data, CCspEngineHandle engine,
100+
CCspDateTime start_time, CCspDateTime end_time)
101+
{
102+
ManagedOutputState* state = (ManagedOutputState*)user_data;
103+
(void)engine;
104+
(void)start_time;
105+
(void)end_time;
106+
107+
printf(" [%s/%s] Output adapter started\n",
108+
state->shared->name, state->topic);
109+
state->messages_sent = 0;
110+
}
111+
112+
static void managed_output_stop(void* user_data)
113+
{
114+
ManagedOutputState* state = (ManagedOutputState*)user_data;
115+
printf(" [%s/%s] Output adapter stopped. Messages sent: %d\n",
116+
state->shared->name, state->topic, state->messages_sent);
117+
}
118+
119+
static void managed_output_execute(void* user_data, CCspEngineHandle engine,
120+
CCspInputHandle input)
121+
{
122+
ManagedOutputState* state = (ManagedOutputState*)user_data;
123+
124+
if (!ccsp_input_is_valid(input)) {
125+
return;
126+
}
127+
128+
/* Get current engine time */
129+
CCspDateTime now = ccsp_engine_now(engine);
130+
double now_sec = (double)now / 1000000000.0;
131+
132+
/* Get the input type and value */
133+
CCspType type = ccsp_input_get_type(input);
134+
135+
printf(" [%s/%s] t=%.3f -> ", state->shared->name, state->topic, now_sec);
136+
137+
switch (type) {
138+
case CCSP_TYPE_INT64: {
139+
int64_t value;
140+
if (ccsp_input_get_last_int64(input, &value) == CCSP_OK) {
141+
printf("int64: %lld\n", (long long)value);
142+
}
143+
break;
144+
}
145+
case CCSP_TYPE_DOUBLE: {
146+
double value;
147+
if (ccsp_input_get_last_double(input, &value) == CCSP_OK) {
148+
printf("double: %.6f\n", value);
149+
}
150+
break;
151+
}
152+
case CCSP_TYPE_STRING: {
153+
const char* data;
154+
size_t length;
155+
if (ccsp_input_get_last_string(input, &data, &length) == CCSP_OK) {
156+
printf("string: \"%.*s\"\n", (int)length, data);
157+
}
158+
break;
159+
}
160+
default:
161+
printf("(type %d)\n", type);
162+
break;
163+
}
164+
165+
state->messages_sent++;
166+
state->shared->message_count++;
167+
}
168+
169+
static void managed_output_destroy(void* user_data)
170+
{
171+
ManagedOutputState* state = (ManagedOutputState*)user_data;
172+
printf(" [%s/%s] Output adapter destroyed\n",
173+
state->shared->name, state->topic);
174+
free(state);
175+
}
176+
177+
CCspOutputAdapterVTable example_managed_output_adapter_create(
178+
ManagedAdapterState* shared_state,
179+
const char* topic)
180+
{
181+
CCspOutputAdapterVTable vtable = {0};
182+
183+
if (!shared_state) {
184+
return vtable;
185+
}
186+
187+
ManagedOutputState* state = (ManagedOutputState*)malloc(sizeof(ManagedOutputState));
188+
if (!state) {
189+
return vtable;
190+
}
191+
192+
memset(state, 0, sizeof(ManagedOutputState));
193+
state->shared = shared_state;
194+
if (topic) {
195+
strncpy(state->topic, topic, sizeof(state->topic) - 1);
196+
} else {
197+
strncpy(state->topic, "default", sizeof(state->topic) - 1);
198+
}
199+
200+
vtable.user_data = state;
201+
vtable.start = managed_output_start;
202+
vtable.stop = managed_output_stop;
203+
vtable.execute = managed_output_execute;
204+
vtable.destroy = managed_output_destroy;
205+
206+
return vtable;
207+
}

‎cpp/csp/adapters/c/example/ExampleManagedAdapter.h‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
/*
2+
* Example Managed Adapter using the CSP C API
3+
*
4+
* This example demonstrates:
5+
* - Creating an adapter manager that coordinates multiple adapters
6+
* - Managing lifecycle (start/stop) across adapters
7+
* - Status reporting
8+
* - Coordinated output adapters
9+
*
10+
* This is a more realistic example than ExampleOutputAdapter.c,
11+
* showing how real adapters like Kafka or WebSocket would be structured.
12+
*/
13+
14+
#ifndef _IN_CSP_ADAPTERS_C_EXAMPLE_MANAGED_ADAPTER_H
15+
#define _IN_CSP_ADAPTERS_C_EXAMPLE_MANAGED_ADAPTER_H
16+
17+
#include <csp/engine/c/AdapterManager.h>
18+
19+
#ifdef __cplusplus
20+
extern "C" {
21+
#endif
22+
23+
/*
24+
* ManagedAdapterState - Shared state for managed adapters
25+
*
26+
* In a real adapter (like Kafka), this would contain:
27+
* - Connection handles
28+
* - Configuration
29+
* - Thread pools
30+
* - Message buffers
31+
*/
32+
typedef struct ManagedAdapterState {
33+
char name[64];
34+
int is_started;
35+
int message_count;
36+
CCspAdapterManagerHandle manager;
37+
} ManagedAdapterState;
38+
39+
/*
40+
* ManagedOutputState - State for a single output adapter in the manager
41+
*/
42+
typedef struct ManagedOutputState {
43+
ManagedAdapterState* shared;
44+
char topic[64]; /* e.g., Kafka topic name */
45+
int messages_sent;
46+
} ManagedOutputState;
47+
48+
/*
49+
* example_managed_adapter_create - Create managed adapter VTable
50+
*
51+
* Creates an adapter manager that can coordinate multiple output adapters.
52+
* Similar to how KafkaAdapterManager works.
53+
*
54+
* Parameters:
55+
* name - Name for this adapter manager instance
56+
*
57+
* Returns:
58+
* VTable for use with ccsp_adapter_manager_extern_create
59+
*/
60+
CCspAdapterManagerVTable example_managed_adapter_create(const char* name);
61+
62+
/*
63+
* example_managed_output_adapter_create - Create output adapter for manager
64+
*
65+
* Creates an output adapter that works with the managed adapter.
66+
*
67+
* Parameters:
68+
* shared_state - Pointer to ManagedAdapterState from the manager
69+
* topic - Topic/channel name for this output
70+
*
71+
* Returns:
72+
* VTable for use with ccsp_adapter_manager_create_output_adapter
73+
*/
74+
CCspOutputAdapterVTable example_managed_output_adapter_create(
75+
ManagedAdapterState* shared_state,
76+
const char* topic);
77+
78+
#ifdef __cplusplus
79+
}
80+
#endif
81+
82+
#endif /* _IN_CSP_ADAPTERS_C_EXAMPLE_MANAGED_ADAPTER_H */

0 commit comments

Comments
 (0)