mirror of
https://github.com/haiwen/ccnet-server.git
synced 2025-05-01 03:53:20 +00:00
152 lines
4.2 KiB
C
152 lines
4.2 KiB
C
/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
|
|
|
|
#include <config.h>
|
|
|
|
#include <stdio.h>
|
|
|
|
#include "ccnet-client.h"
|
|
#include "mqclient-proc.h"
|
|
|
|
#define SC_MSG "300"
|
|
#define SC_UNSUBSCRIBE "301"
|
|
|
|
enum {
|
|
INIT,
|
|
REQUEST_SENT,
|
|
READY
|
|
};
|
|
|
|
enum {
|
|
RECV_MSG_SIG,
|
|
LAST_SIGNAL
|
|
};
|
|
|
|
static guint signals[LAST_SIGNAL] = { 0 };
|
|
|
|
static int mq_client_start (CcnetProcessor *processor, int argc, char **argv);
|
|
|
|
static void handle_response (CcnetProcessor *processor,
|
|
char *code, char *code_msg,
|
|
char *content, int clen);
|
|
|
|
|
|
G_DEFINE_TYPE (CcnetMqclientProc, ccnet_mqclient_proc, CCNET_TYPE_PROCESSOR)
|
|
|
|
static void
|
|
ccnet_mqclient_proc_class_init (CcnetMqclientProcClass *klass)
|
|
{
|
|
CcnetProcessorClass *proc_class = CCNET_PROCESSOR_CLASS (klass);
|
|
|
|
proc_class->start = mq_client_start;
|
|
proc_class->handle_response = handle_response;
|
|
proc_class->name = "mqclient-proc";
|
|
|
|
signals[RECV_MSG_SIG] =
|
|
g_signal_new ("recv-msg", CCNET_TYPE_MQCLIENT_PROC ,
|
|
G_SIGNAL_RUN_LAST,
|
|
0,
|
|
NULL, NULL, /* no accumulator */
|
|
g_cclosure_marshal_VOID__OBJECT,
|
|
G_TYPE_NONE, 1, CCNET_TYPE_MESSAGE);
|
|
}
|
|
|
|
static void
|
|
ccnet_mqclient_proc_init (CcnetMqclientProc *processor)
|
|
{
|
|
CCNET_PROCESSOR (processor)->state = INIT;
|
|
}
|
|
|
|
static int
|
|
mq_client_start (CcnetProcessor *processor, int argc, char **argv)
|
|
{
|
|
GString *buf;
|
|
int i;
|
|
|
|
buf = g_string_new ("mq-server");
|
|
for (i = 0; i < argc; ++i) {
|
|
g_string_append (buf, " ");
|
|
g_string_append (buf, argv[i]);
|
|
}
|
|
|
|
ccnet_client_send_request (processor->session, REQUEST_ID (processor->id),
|
|
buf->str);
|
|
processor->state = REQUEST_SENT;
|
|
|
|
g_string_free (buf, TRUE);
|
|
return 0;
|
|
}
|
|
|
|
static void handle_response (CcnetProcessor *processor,
|
|
char *code, char *code_msg,
|
|
char *content, int clen)
|
|
{
|
|
CcnetMqclientProc *proc = (CcnetMqclientProc *) processor;
|
|
CcnetMessage *msg;
|
|
|
|
switch (processor->state) {
|
|
case REQUEST_SENT:
|
|
if (code[0] != '2') {
|
|
g_warning ("receive bad response: %s %s\n", code, code_msg);
|
|
ccnet_processor_done (processor, FALSE);
|
|
return;
|
|
}
|
|
|
|
processor->state = READY;
|
|
break;
|
|
case READY:
|
|
if (code[0] != '2' && code[0] != '3') {
|
|
g_warning ("receive bad response: %s %s\n", code, code_msg);
|
|
return;
|
|
}
|
|
|
|
/* message notification. */
|
|
if (code[0] == '3' && code[2] == '0') {
|
|
msg = ccnet_message_from_string (content, clen);
|
|
if (!msg) {
|
|
g_warning ("Wrong message format.\n");
|
|
ccnet_processor_done (processor, FALSE);
|
|
break;
|
|
}
|
|
if (proc->message_got_cb)
|
|
proc->message_got_cb (msg, proc->cb_data);
|
|
g_signal_emit (proc, signals[RECV_MSG_SIG], 0, msg);
|
|
ccnet_message_free (msg);
|
|
}
|
|
|
|
break;
|
|
default:
|
|
break;
|
|
}
|
|
}
|
|
|
|
void ccnet_mqclient_proc_set_message_got_cb (CcnetMqclientProc *processor,
|
|
MessageGotCB callback,
|
|
void *cb_data)
|
|
{
|
|
processor->message_got_cb = callback;
|
|
processor->cb_data = cb_data;
|
|
}
|
|
|
|
void ccnet_mqclient_proc_put_message (CcnetMqclientProc *proc,
|
|
CcnetMessage *message)
|
|
{
|
|
CcnetProcessor *processor = (CcnetProcessor *) proc;
|
|
GString *msg_buf;
|
|
|
|
msg_buf = g_string_new (NULL);
|
|
|
|
ccnet_message_to_string_buf (message, msg_buf);
|
|
ccnet_client_send_update (processor->session, UPDATE_ID(processor->id),
|
|
SC_MSG, NULL, msg_buf->str, msg_buf->len+1);
|
|
g_string_free (msg_buf, TRUE);
|
|
}
|
|
|
|
void ccnet_mqclient_proc_unsubscribe_apps (CcnetMqclientProc *proc)
|
|
{
|
|
CcnetProcessor *processor = (CcnetProcessor *) proc;
|
|
|
|
ccnet_client_send_update (processor->session, UPDATE_ID(processor->id),
|
|
SC_UNSUBSCRIBE, NULL, NULL, 0);
|
|
ccnet_processor_done (processor, TRUE);
|
|
}
|