1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
|
#include "coordinator.h"
#include "pthread.h"
#include "nanomsg/nn.h"
#include "nanomsg/survey.h"
#include "json-c/json.h"
#include <stdio.h>
#include <string.h>
#include "sleep.h"
static pthread_t coordinator_thread;
static int coordinator_sock;
static int coordinator_must_terminate = 0;
static void *coordinator_func( void *thread_data )
{
char *control = (char *)thread_data;
coordinator_sock = nn_socket( AF_SP, NN_RESPONDENT );
(void)nn_connect( coordinator_sock, control );
printf( "coordinator connected to %s\n", control );
sleep( 1 );
while( !coordinator_must_terminate ) {
printf( "coordinator idle: %d\n", coordinator_must_terminate );
sleep( 1 );
char *answer = NULL;
int bytes = nn_recv( coordinator_sock, &answer, NN_MSG, 0 );
if( bytes == ETIMEDOUT ) continue;
if( coordinator_must_terminate ) continue;
if( bytes >= 0 ) {
printf( "coordinator received: %s\n", answer );
nn_freemsg( answer );
json_object *obj = json_object_new_object( );
json_object *op = json_object_new_string( "register" );
json_object_object_add( obj, "op", op );
json_object *role = json_object_new_string( "coordinator" );
json_object_object_add( obj, "role", role );
const char *msg = json_object_to_json_string( obj );
int msg_size = strlen( msg ) + 1;
printf( "coordinator send: %s\n", msg );
bytes = nn_send( coordinator_sock, msg, msg_size, 0 );
if( bytes != msg_size ) {
fprintf( stderr, "ERROR: truncated message!" );
}
json_object_put( obj );
}
}
(void)nn_shutdown( coordinator_sock, 0 );
puts( "coordinator disconnected" );
return NULL;
}
int coordinator_init( const char *control )
{
pthread_attr_t attr;
int res;
coordinator_must_terminate = 0;
res = pthread_attr_init( &attr );
if( res != 0 ) {
return 1;
}
res = pthread_create( &coordinator_thread, &attr, coordinator_func, (void *)control );
if( res != 0 ) {
return 1;
}
return 0;
}
void coordinator_terminate( )
{
coordinator_must_terminate = 1;
nn_term( );
}
int coordinator_free( )
{
void *result;
int res;
res = pthread_join( coordinator_thread, &result );
if( res != 0 ) {
return 1;
}
return 0;
}
|
