SimulationCore2/Redis/redis_io.c

#include <stdio.h>
#include <string.h>
#include <stdlib.h>
#include <signal.h>
#include "../include/hiredis.h"
#include "../include/async.h"
#include "../include/adapters/libevent.h"
redisAsyncContext * asyncContext;
redisContext *context;
char *result,*key,*pass,*channel_in,*channel_out;

extern void test();
extern void setTongLever(int *v);
extern void ButtonPress_Slips();
extern void ToggleValve(int *valve_number);

char *clone(char *s)
{
    char *r = malloc(strlen(s)+1);

}
void initConnection(char *address, int *port,char * password,char *datakey,int *returnValue)
{
    context = redisConnect(address,*port);
    asyncContext = redisAsyncConnect(address, *port);

    if (context == NULL || context->err || asyncContext==NULL || asyncContext->err)
    {
        if (context) {
            printf("Error: %s\n", context->errstr);
        } else {
            printf("Can't allocate redis context\n");
        }
        (*returnValue) = -1;
        if (asyncContext) {
            printf("Error: %s\n", asyncContext->errstr);
        } else {
            printf("Can't allocate Redis Async Context\n");
        }
        (*returnValue) = -1;
        return;
    }
    printf("Connection Stablished to %s\n",address);
    if(strlen(password)>0)
    {
        // printf("Authenticating with password %s (len={%zu})",password,strlen(password));
        redisReply *reply= redisCommand(context, "AUTH %s", password);
        redisAsyncCommand(asyncContext, NULL,NULL,"AUTH %s", password);
        if (reply->type == REDIS_REPLY_ERROR) {
            printf("Authentication failed.\n");
            (*returnValue) = -1;
            return;
        }
        else {
            printf("Authentication is done.\n");
        }
        freeReplyObject(reply);
        (*returnValue) = 1;
    }
    key = malloc((sizeof(char) * (strlen(datakey)+1)));
    pass = malloc((sizeof(char) * (strlen(password)+1)));
    channel_in = malloc((sizeof(char) * (strlen(datakey)+7)));
    channel_out = malloc((sizeof(char) * (strlen(datakey)+8)));
    strcpy(key,datakey);
    strcpy(pass,password);
    strcpy(channel_in,datakey);
    strcpy(channel_out,datakey);
    strcat(channel_in,".ch_in");
    strcat(channel_out,".ch_out");
}
void setData(char *part, char *data)
{
    redisReply *reply;
    // printf("%zu chars written\n",strlen(data));
    reply = redisCommand(context, "SET %s.%s %s",key,part,data);
    freeReplyObject(reply);
}


char *getData(int *len)
{
    redisReply *reply;
    reply = redisCommand(context, "GET %s.in",key);
    result = (char*) malloc(*len);
    strcpy(result,reply->str);
    int paddingLength = (*len) - strlen(reply->str);
    if (paddingLength > 0) {
        memset(result + strlen(reply->str), ' ', paddingLength-1);
    }
    *len = strlen(reply->str);
    freeReplyObject(reply);
    return result;
}

void deallocData()
{
    free(result);
    free(context);
    free(key);
}

void onMessage(redisAsyncContext * c, void *reply, void * privdata) {
    int j,v;
    redisReply * r = reply;
    if (reply == NULL) return;

    printf("got a message of type: %i\n", r->type);
    
    if (r->type == REDIS_REPLY_ARRAY && r->elements==3) {
        if(strcmp(r->element[0]->str,"message")==0)
        if(strcmp(r->element[1]->str,channel_in)==0)
        {
            // char *message = malloc(srlen(r->element[2]->str));
            // strcpy(message,r->element[2]->str);
            char delimiter = ',';
            char *chptr = strchr(r->element[2]->str, delimiter);
            char *fn = strtok(r->element[2]->str,&delimiter);
            char *parameter = malloc(20);
            parameter = strtok(NULL,&delimiter);
            printf("calling %s\n",fn);
            if(strcmp(fn,"test")==0)
                test();
            else if(strcmp(fn,"TONG_LEVER_MAKEUP")==0){
                v = -1;
                setTongLever(&v);
            }
            else if(strcmp(fn,"TONG_LEVER_BREAKOUT")==0){
                v=1;
                setTongLever(&v);
            }else if(strcmp(fn,"BUTTON_PRESS_SLIPS")==0){
                ButtonPress_Slips();
            }else if(strcmp(fn,"ChangeValve")==0){
                int valve_number = atoi(parameter);
                ToggleValve(&valve_number);
            // }else if(strcmp(fn,"KellyConnected")==0){
            //     KellyConnected();
            // }else if(strcmp(fn,"KellyDisconnected")==0){
            //     KellyDisconnected();
            }else
                printf("message: %s\n",r->element[2]->str);
        }
    }
    else{
        for(int i=0;i<r->elements;i++)
            printf("%d) %s",i,r->element[i]->str);
    }
}



void listenTochannel()
{
    printf("Listening To channel (C)\n");
    signal(SIGPIPE, SIG_IGN);
    struct event_base * base = event_base_new();
    redisLibeventAttach(asyncContext, base);
    redisAsyncCommand(asyncContext, onMessage, NULL, "SUBSCRIBE %s",channel_in);
    printf("Subscribed to channel %s\n",channel_in);
    event_base_dispatch(base);
}

void publishMessageToChannel(const char *message) {
    redisReply *reply;
    reply = redisCommand(context, "PUBLISH %s %s", channel_out, message);
    
    if (reply == NULL) {
        printf("Error: publishMessageToChannel failed.\n");
    }
    freeReplyObject(reply);
}