recv.c

Go to the documentation of this file.

 
#include <inttypes.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
 
#ifndef _WIN32
    #include <pthread.h>
    #include <unistd.h>
#else /* _WIN32 */
    #include <process.h>
    #include <windows.h>
#endif /* _WIN32 */
 
#include <porter.h>
#include <console-util.h>

static volatile int g_running = 1;
 
#ifndef _WIN32
static void sig_handler(int sig)
{
    (void)sig;
    g_running = 0;
    printf("\n終了中...\n");
    fflush(stdout);
}
#else /* _WIN32 */
static BOOL WINAPI console_ctrl_handler(DWORD type)
{
    switch (type)
    {
    case CTRL_C_EVENT:
    case CTRL_BREAK_EVENT:
    case CTRL_CLOSE_EVENT:
    case CTRL_SHUTDOWN_EVENT:
        g_running = 0;
        printf("\n終了中...\n");
        fflush(stdout);
        return TRUE;
    }
 
    return FALSE;
}
#endif /* _WIN32 */

static void on_recv(int64_t service_id, PotrPeerId peer_id, PotrEvent event, const void *data, size_t len)
{
    char buf[POTR_MAX_PAYLOAD + 1];
    size_t copy_len;
 
    (void)peer_id;
    switch (event)
    {
    case POTR_EVENT_CONNECTED:
        printf("[サービス %" PRId64 "] 接続確立\n", service_id);
        fflush(stdout);
        break;
 
    case POTR_EVENT_DISCONNECTED:
        printf("[サービス %" PRId64 "] 切断検知\n", service_id);
        fflush(stdout);
        break;
 
    case POTR_EVENT_DATA:
    default:
        if (len < POTR_MAX_PAYLOAD)
        {
            copy_len = len;
        }
        else
        {
            copy_len = POTR_MAX_PAYLOAD;
        }
        memcpy(buf, data, copy_len);
        buf[copy_len] = '\0';
        printf("[サービス %" PRId64 "] 受信 (%zu バイト): %s\n", service_id, len, buf);
        fflush(stdout);
        break;
    }
}

static int parse_log_level(const char *str, PotrLogLevel *out)
{
    static const struct
    {
        const char *name;
        PotrLogLevel level;
        uint32_t _pad;
    } tbl[] = {
        {"DEBUG", POTR_TRACE_VERBOSE, 0U}, {"INFO", POTR_TRACE_INFO, 0U},
        {"WARN", POTR_TRACE_WARNING, 0U},   {"ERROR", POTR_TRACE_ERROR, 0U}, {"FATAL", POTR_TRACE_CRITICAL, 0U},
    };
    char upper[16];
    size_t i;
    size_t j;
 
    for (j = 0; j < sizeof(upper) - 1U && str[j] != '\0'; j++)
    {
        upper[j] = (str[j] >= 'a' && str[j] <= 'z') ? (char)(str[j] - ('a' - 'A')) : str[j];
    }
    upper[j] = '\0';
 
    for (i = 0; i < sizeof(tbl) / sizeof(tbl[0]); i++)
    {
        if (strcmp(upper, tbl[i].name) == 0)
        {
            *out = tbl[i].level;
            return 1;
        }
    }
    return 0;
}
 
/* ============================================================ */
/* unicast_bidir 送信スレッド                                     */
/* ============================================================ */

typedef struct
{
    PotrHandle handle;
    volatile int *running;
} BidirSendCtx;

static int read_line(char *buf, size_t size)
{
    if (fgets(buf, (int)size, stdin) == NULL)
    {
        return 0;
    }
    buf[strcspn(buf, "\n")] = '\0';
    return 1;
}
 
#ifndef _WIN32
typedef pthread_t BidirThread;

static void *bidir_send_thread_func(void *arg)
#else  /* _WIN32 */
typedef HANDLE BidirThread;

static unsigned __stdcall bidir_send_thread_func(void *arg)
#endif /* _WIN32 */
{
    BidirSendCtx *ctx = (BidirSendCtx *)arg;
    char msg_buf[POTR_MAX_MESSAGE_SIZE + 2U];
    char ans_buf[8];
    size_t msg_len;
    int compress;
    const char *compress_label;
 
    while (*ctx->running)
    {
        printf("\nメッセージ> ");
        fflush(stdout);
 
        if (!read_line(msg_buf, sizeof(msg_buf)))
        {
            break; /* EOF またはシグナル割り込み */
        }
 
        msg_len = strlen(msg_buf);
        if (msg_len == 0U)
        {
            break; /* 空行で送信終了 */
        }
 
        printf("圧縮送信しますか？ [y/N]> ");
        fflush(stdout);
 
        compress = 0;
        if (read_line(ans_buf, sizeof(ans_buf)))
        {
            if (ans_buf[0] == 'y' || ans_buf[0] == 'Y')
            {
                compress = 1;
            }
        }
 
        compress_label = compress ? " [圧縮あり]" : "";
        printf("送信中: \"%s\" (%zu バイト)%s\n", msg_buf, msg_len, compress_label);
 
        if (potrSend(ctx->handle, POTR_PEER_NA, msg_buf, msg_len,
                     (compress ? POTR_SEND_COMPRESS : 0) | POTR_SEND_BLOCKING) != POTR_SUCCESS)
        {
            fprintf(stderr, "エラー: 送信に失敗しました。\n");
            break;
        }
 
        printf("送信完了。");
 
        printf(" 続けて送信しますか？ [Y/n]> ");
        fflush(stdout);
 
        if (!read_line(ans_buf, sizeof(ans_buf)))
        {
            break;
        }
 
        if (ans_buf[0] == 'n' || ans_buf[0] == 'N')
        {
            break;
        }
    }
 
    *ctx->running = 0; /* 送信終了時に受信ループも停止させる */
 
#ifndef _WIN32
    return NULL;
#else  /* _WIN32 */
    return 0U;
#endif /* _WIN32 */
}

static int start_bidir_send_thread(BidirThread *thread, BidirSendCtx *ctx)
{
#ifndef _WIN32
    return pthread_create(thread, NULL, bidir_send_thread_func, ctx) == 0;
#else  /* _WIN32 */
    uintptr_t h = _beginthreadex(NULL, 0U, bidir_send_thread_func, ctx, 0U, NULL);
    if (h == 0U)
    {
        return 0;
    }
    *thread = (HANDLE)h;
    return 1;
#endif /* _WIN32 */
}

static void join_bidir_send_thread(BidirThread thread)
{
#ifndef _WIN32
    pthread_cancel(thread); /* fgets でブロック中のスレッドを中断する */
    pthread_join(thread, NULL);
#else  /* _WIN32 */
    WaitForSingleObject(thread, INFINITE);
    CloseHandle(thread);
#endif /* _WIN32 */
}

int main(int argc, char *argv[])
{
    const char *config_path;
    int64_t service_id;
    PotrHandle handle;
    int i;
    PotrLogLevel log_level = POTR_TRACE_NONE;
    int log_level_set = 0;
    PotrType svc_type;
    int is_bidir;
    BidirSendCtx bidir_ctx;
    BidirThread bidir_thread = 0;
    int bidir_started = 0;
 
    /* コンソール UTF-8 ヘルパーを初期化する */
    console_init();
 
    /* オプション解析 */
    for (i = 1; i < argc; i++)
    {
        if (strcmp(argv[i], "-l") == 0)
        {
            if (i + 1 >= argc)
            {
                fprintf(stderr, "エラー: -l オプションにレベルを指定してください。\n");
                fprintf(stderr, "使用方法: %s [-l <level>] <config_path> <service_id>\n", argv[0]);
                console_dispose();
                return EXIT_FAILURE;
            }
            i++;
            if (!parse_log_level(argv[i], &log_level))
            {
                fprintf(stderr,
                        "エラー: 不明なログレベル \"%s\"。"
                        "TRACE/DEBUG/INFO/WARN/ERROR/FATAL のいずれかを指定してください。\n",
                        argv[i]);
                console_dispose();
                return EXIT_FAILURE;
            }
            log_level_set = 1;
        }
        else
        {
            break; /* 最初の非オプション引数で停止 */
        }
    }
 
    /* positional 引数チェック */
    if (argc - i < 2)
    {
        fprintf(stderr, "使用方法: %s [-l <level>] <config_path> <service_id>\n", argv[0]);
        fprintf(stderr, "  -l <level>  ログレベル (TRACE/DEBUG/INFO/WARN/ERROR/FATAL)\n");
        fprintf(stderr, "例: %s porter-services.conf 10\n", argv[0]);
        fprintf(stderr, "例: %s -l INFO porter-services.conf 10\n", argv[0]);
        console_dispose();
        return EXIT_FAILURE;
    }
 
    config_path = argv[i];
    service_id = (int64_t)strtoll(argv[i + 1], NULL, 10);
 
    /* ロガー設定 (stderr 出力、ファイルなし) */
    if (log_level_set)
    {
        if (potrLogConfig(log_level, NULL, 1) != POTR_SUCCESS)
        {
            fprintf(stderr, "エラー: ロガーの設定に失敗しました。\n");
            console_dispose();
            return EXIT_FAILURE;
        }
    }
 
#ifndef _WIN32
    signal(SIGINT, sig_handler);
#else  /* _WIN32 */
    SetConsoleCtrlHandler(console_ctrl_handler, TRUE);
#endif /* _WIN32 */
 
    printf("サービス %" PRId64 " を開いています... (設定: %s)\n", service_id, config_path);
    fflush(stdout);
 
    /* サービス種別を取得して unicast_bidir かどうか判定する */
    is_bidir = 0;
    if (potrGetServiceType(config_path, service_id, &svc_type) == POTR_SUCCESS && svc_type == POTR_TYPE_UNICAST_BIDIR)
    {
        is_bidir = 1;
    }
 
    if (potrOpenServiceFromConfig(config_path, service_id, POTR_ROLE_RECEIVER, on_recv, &handle) != POTR_SUCCESS)
    {
        fprintf(stderr, "エラー: サービス %" PRId64 " を開けませんでした。\n", service_id);
        console_dispose();
        return EXIT_FAILURE;
    }
 
    if (is_bidir)
    {
        printf("双方向モード (unicast_bidir)。\n");
        printf("メッセージを入力して送信できます (空行または Ctrl+D で送信終了)。\n");
        fflush(stdout);
        bidir_ctx.handle = handle;
        bidir_ctx.running = &g_running;
        if (start_bidir_send_thread(&bidir_thread, &bidir_ctx))
        {
            bidir_started = 1;
        }
        else
        {
            fprintf(stderr, "警告: 送信スレッドの起動に失敗しました。受信専用モードで動作します。\n");
        }
    }
 
    printf("受信待機中... (Ctrl+C で終了)\n");
    fflush(stdout);
 
    while (g_running)
    {
#ifndef _WIN32
        usleep(100000);
#else  /* _WIN32 */
        Sleep(100);
#endif /* _WIN32 */
    }
 
    if (bidir_started)
    {
        join_bidir_send_thread(bidir_thread);
    }
 
    potrCloseService(handle);
    printf("終了しました。\n");
    fflush(stdout);
    console_dispose();
    return EXIT_SUCCESS;
}