/*
 * redis.c  (C) SOFNEC
 *
 */
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#include "hiredis.h"
#include "putil.h"

#include <azst.h>

//#define DEBUG
//#define BACKWARD_COMPAT

#define AZREDIS_VERSION   "0.9"

#define DEFAULT_MAX_REPLY_NUM   1000
#define MAX_COMMAND_ARGS_NUM     200

#define INVALID_ARG      AZ_ERROR(9)

#define CONV_DEFAULT    0
#define CONV_TERM       1
#define CONV_ATOM       2

static pred P1_redis_is_disconnected(Frame *Env);
static pred P4_redis_connect(Frame *Env);
static pred P1_redis_free_context(Frame *Env);
static pred P4_redis_command(Frame *Env);
static pred P4_redis_command_to_term(Frame *Env);
static pred P4_redis_command_to_atom_list(Frame *Env);
static pred P4_redis_command_hate_integer(Frame *Env);
static pred P2_redis_next_val(Frame *Env);
static pred P1_redis_set_max_reply_num(Frame *Env);
//static pred P1_redis_set_reply_value_atom(Frame *Env);
static pred P1_redis_version(Frame *Env);
static pred P4_redis_string_replace(Frame *Env);

static BASEINT REDIS_REPLY_STRING_ATOM;
static BASEINT REDIS_REPLY_ARRAY_ATOM;
static BASEINT REDIS_REPLY_INTEGER_ATOM;
static BASEINT REDIS_REPLY_NIL_ATOM;
static BASEINT REDIS_REPLY_STATUS_ATOM;
static BASEINT REDIS_REPLY_ERROR_ATOM;
static BASEINT REDIS_REPLY_UNKNOWN_ATOM;
static BASEINT REDIS_VERSION_ATOM;

static BASEINT ON_ATOM;
static BASEINT OFF_ATOM;

static st_table* ContextTable;

static int MaxReplyNum = DEFAULT_MAX_REPLY_NUM;
static int ReplyArrayStart;
static redisReply* SaveReply;
static int SaveHateInteger;
static int SaveConvType;


#ifdef WIN32
  __declspec(dllexport) extern int initiate_redis(void)
#else
  extern int initiate_redis(void)
#endif
{
  char buf[256];

  if (ContextTable != 0)
    st_free_table(ContextTable);

  ContextTable = st_init_numtable();

  put_bltn("redis_connect\0",               4, P4_redis_connect);
  put_bltn("redis_free_context\0",          1, P1_redis_free_context);
  put_bltn("redis_command\0",               4, P4_redis_command);
  put_bltn("redis_command_to_term\0",       4, P4_redis_command_to_term);
  put_bltn("redis_command_to_atom_list\0",  4, P4_redis_command_to_atom_list);
  put_bltn("redis_command_hate_integer\0",  4, P4_redis_command_hate_integer);
  put_bltn("redis_next_val\0",              2, P2_redis_next_val);
  put_bltn("redis_set_max_reply_num\0",     1, P1_redis_set_max_reply_num);
  //put_bltn("redis_set_reply_value_atom\0",  1, P1_redis_set_reply_value_atom);
  put_bltn("redis_version\0",               1, P1_redis_version);
  put_bltn("redis_string_replace\0",        4, P4_redis_string_replace);

  /* !!! redis_is_disconnected() must be called after redis_command() fail. */
  put_bltn("redis_is_disconnected\0",      1, P1_redis_is_disconnected);

  REDIS_REPLY_STRING_ATOM  = PutAtom((Frame* )NULL, "redis_REPLY_STRING");
  REDIS_REPLY_ARRAY_ATOM   = PutAtom((Frame* )NULL, "redis_REPLY_ARRAY");
  REDIS_REPLY_INTEGER_ATOM = PutAtom((Frame* )NULL, "redis_REPLY_INTEGER");
  REDIS_REPLY_NIL_ATOM     = PutAtom((Frame* )NULL, "redis_REPLY_NIL");
  REDIS_REPLY_STATUS_ATOM  = PutAtom((Frame* )NULL, "redis_REPLY_STATUS");
  REDIS_REPLY_ERROR_ATOM   = PutAtom((Frame* )NULL, "redis_REPLY_ERROR");
  REDIS_REPLY_UNKNOWN_ATOM = PutAtom((Frame* )NULL, "redis_REPLY_UNKNOWN");

  ON_ATOM  = PutAtom((Frame* )NULL, "on");
  OFF_ATOM = PutAtom((Frame* )NULL, "off");

  sprintf(buf, "redis-ext: %s, hiredis: %d.%d.%d", AZREDIS_VERSION,
          HIREDIS_MAJOR, HIREDIS_MINOR, HIREDIS_PATCH);
  REDIS_VERSION_ATOM = PutAtom((Frame* )NULL, buf);

  return 1;
}

static int
get_term_str(Frame* Env, TERM* t, char** s)
{
  int len;

  REALVALUE(t);
  len = az_term_to_cstring_length(Env, t);
  *s = malloc(len + 1);
  if (*s == 0) return -1;

  (void )az_term_to_cstring(Env, t, *s, len + 1);

  return len;
}

static int
term_to_string(Frame* Env, TERM* t, char** rs)
{
#ifdef BACKWARD_COMPAT
  AZ_EXTERN int B2_term_string(Frame *Env, TERM *arg1, TERM *arg2);
#endif

  TERM* x;
  int r;
  int len;

/* 
20151001 Telly Found Serious Mistake!
   GVCLEAR means "Global Var Clear"
   Never use to LocalStack or auto C area
*/

#if 0	
  GVCLEAR(next_var_cell);
  x = next_var_cell++;
#else
  x = next_var_cell;
  MakeUndef(Env);
#endif

  r = B2_term_string(Env, t, x);
  //fprintf(stderr, "r: %d\r\n", r);
  if (r == 0) {
    next_var_cell = x;
    return -1;
  }

  len = get_term_str(Env, x, rs);
  //fprintf(stderr, "len: %d\r\n", len);
  next_var_cell = x;
  if (len < 0) {
    return -2;
  }

  return len;
}

static int
string_to_term(Frame* Env, TERM* t, char* s)
{
#ifdef BACKWARD_COMPAT
  AZ_EXTERN struct _Stream* bufferstream;
  AZ_EXTERN TERM* gvar_top;

  AZ_EXTERN void Stream_ClearUngetBuff(struct _Stream *stream);
  AZ_EXTERN void Stream_ResetBuffer(struct _Stream *stream);
  AZ_EXTERN int  Stream_buf_put(int c, struct _Stream *s);
  AZ_EXTERN int  Read_ReadGeneric(Frame *Env, struct _Stream *stream);
#endif

  char* p;
  char* end;
  TERM *sv, *unifyp;
  int r;
#ifndef BACKWARD_COMPAT
  int v;
#endif

  end = s + strlen(s);

  Stream_ClearUngetBuff(bufferstream);
  Stream_ResetBuffer(bufferstream);

  p = s;
  while (p < end) {
    Stream_buf_put((int )*p, bufferstream);

    p++;
  }

  Stream_buf_put(' ', bufferstream);
  Stream_buf_put('.', bufferstream);
  Stream_buf_put(0,   bufferstream);
  Stream_ResetBuffer(bufferstream);

  PUSH_STACK(t);
  sv = next_var_cell;
#ifdef BACKWARD_COMPAT
  r = Read_ReadGeneric(Env, bufferstream);
#else
  r = Read_ReadGeneric(Env, bufferstream, &v);
#endif
  if (r == 1) {
    unifyp = next_var_cell - 1;
    next_var_cell = sv;
    POP_STACK(t);
    r = unifyE_ex(Env, t, unifyp);
    if (r != 0) return 0;
    return -1;
  }
  next_var_cell = sv - 1;

  return 1;
}

static int
unify_term_or_list(Frame* Env, TERM* t, char* s, int conv_type)
{
  int r;

  switch (conv_type) {
  case CONV_DEFAULT:
    r = az_cstring_to_list(Env, s, strlen(s), t);
    break;

  case CONV_TERM:
    r = string_to_term(Env, t, s);
#ifdef DEBUG
    if (r != 0)
      fprintf(stderr, "string_to_term: %d, {%s}\r\n", r, s);
#endif

    if (r == 1) { // syntax error
      char* conv;

      r = pu_escape_single_quote(s, &conv);
      if (r < 0) return 0;

      r = string_to_term(Env, t, conv);
#ifdef DEBUG
      fprintf(stderr, "string_to_term(conv): %d, {%s} {%s}\r\n", r, s, conv);
#endif
      if (conv != s) free(conv);
      if (r == 1)
        return unify_atom(t, PutAtom(Env, s));
    }

    break;

  case CONV_ATOM:
    r = unify_atom(t, PutAtom(Env, s));
    return r;
    break;

  default:
    return 0; // error
  }

  return r == 0 ? 1 : 0;
}

static int
make_element(redisReply* e, Frame* Env, TERM* t, int hate_integer, int conv_type)
{
  if (e->type == REDIS_REPLY_INTEGER) {
    /* Probably, never come here.*/
    if (hate_integer != 0) {
      char buf[30];
      sprintf(buf, "%lld", e->integer);
      if (unify_term_or_list(Env, t, buf, conv_type) == 0)
        return -1;
    }
    else {
      if (! UnifyInt(t, (BASEINT )(e->integer)))
        return -1;
    }
  }
  else {
    if (e->len == 0) {
      if (UnifyAtom(t, ATOM_NIL) == 0)
        return -1;
    }
    else {
      if (unify_term_or_list(Env, t, e->str, conv_type) == 0)
        return -1;
    }
  }

  return 0;
}

static int
make_list_from_reply_array(Frame* Env, TERM* t, redisReply* r,
                           int* astart, int max_num,
                           int hate_integer, int conv_type)
{
  int cont;
  int len;
  int start;

  start = *astart;
  len = r->elements - start;
  if (len < 0) len = 0;

  if (len > max_num) {
    cont = len - max_num;
    len = max_num;
  }
  else {
    cont = 0;
  }

  if (len == 0) {
    if (UnifyAtom(t, ATOM_NIL) == 0) {
#ifdef DEBUG
      fprintf(stderr, "make_list_from_reply_array() fail. (1)\n");
#endif
      return -1;
    }
  }
  else {
    int i;
    int rec_size, gvar_size;
    TERM *List_top, *x;

    rec_size = len;
    gvar_size = (BASEINT )(gvar_bottom - next_gvar_cell)/2;	
    if (gvar_size < rec_size) {
      P0_s_greclaim(Env);
      gvar_size = (BASEINT)(gvar_bottom - next_gvar_cell)/2;	
      if (gvar_size < rec_size) {
#ifdef DEBUG
        fprintf(stderr, "make_list_from_reply_array() fail. (2)\n");
#endif
        return -2;
      }
    }

    List_top = next_gvar_cell;

    for (i = 0; i < len; i++) {
      SETTAG(next_gvar_cell, list_tag);
      BODY(next_gvar_cell) = next_gvar_cell+1;
      next_gvar_cell++;

      GVCLEAR(next_gvar_cell);
      next_gvar_cell++;
    }
    SETTAG(next_gvar_cell, atom_tag);
    INT_BODY(next_gvar_cell++) = ATOM_NIL;

    i = 0;
    x = List_top;
    while (IsCons(x)) {
      redisReply* e = r->element[start + i];
      REALVALUE(x);
      x = BODY(x);
      if (make_element(e, Env, x, hate_integer, conv_type) != 0) {
#ifdef DEBUG
        fprintf(stderr, "make_list_from_reply_array() fail. (3)\n");
#endif
        return -3;
      }

      x++;
      i++;
    }

    if (unify(List_top, t) == 0) {
#ifdef DEBUG
      fprintf(stderr, "make_list_from_reply_array() fail. (4)\n");
#endif
      return -4;
    }
  }

  *astart = start + len;
  return cont;
}

static int
make_one_element_list(Frame* Env, TERM* t, char* s)
{
  int len;
  int i;
  int rec_size, gvar_size;
  TERM *List_top;
  BASEINT a;

  len = 1;

  a = PutAtom(Env, s);

  rec_size = len;
  gvar_size = (BASEINT )(gvar_bottom - next_gvar_cell)/2;
  if (gvar_size < rec_size) {
    P0_s_greclaim(Env);
    gvar_size = (BASEINT)(gvar_bottom - next_gvar_cell)/2;
    if (gvar_size < rec_size) {
      return -2;
    }
  }

  List_top = next_gvar_cell;

  for (i = 0; i < len; i++) {
    SETTAG(next_gvar_cell, list_tag);
    BODY(next_gvar_cell) = next_gvar_cell+1;
    next_gvar_cell++;

    GVCLEAR(next_gvar_cell);
    if (UnifyAtom(next_gvar_cell, a) == 0) {
      next_gvar_cell = List_top;
      return -1;
    }
    next_gvar_cell++;
  }
  SETTAG(next_gvar_cell, atom_tag);
  INT_BODY(next_gvar_cell++) = ATOM_NIL;

  if (unify(List_top, t) == 0) {
    next_gvar_cell = List_top;
    return -4;
  }

  return 0;
}

static BASEINT
type_to_atom(int type)
{
  BASEINT a;

  switch (type) {
  case REDIS_REPLY_STRING:
    a = REDIS_REPLY_STRING_ATOM;
    break;
  case REDIS_REPLY_ARRAY:
    a = REDIS_REPLY_ARRAY_ATOM;
    break;
  case REDIS_REPLY_INTEGER:
    a = REDIS_REPLY_INTEGER_ATOM;
    break;
  case REDIS_REPLY_NIL:
    a = REDIS_REPLY_NIL_ATOM;
    break;
  case REDIS_REPLY_STATUS:
    a = REDIS_REPLY_STATUS_ATOM;
    break;
  case REDIS_REPLY_ERROR:
    a = REDIS_REPLY_ERROR_ATOM;
    break;
  default:
    a = REDIS_REPLY_UNKNOWN_ATOM;
    break;
  }

  return a;
}

/* ?-redis_connect(+ADDRESS, +PORT, +TIMEOUT(msec.), -CONTEXT). */
static pred
P4_redis_connect(Frame *Env)
{
  int port, t;
  struct timeval timeout;
  redisContext *ctx;
  char buf[AZ_MAX_ATOM_LENGTH + 1];

#ifdef WIN32
  WSADATA data;
  WSAStartup(MAKEWORD(2,0), &data); 
#endif

  ALL_BEGIN(Env);

  if (SaveReply != 0) {
    //freeReplyObject(SaveReply);
    SaveReply = 0;
  }

  Atom2Asciz(GetAtom(PARG(4,0)), buf);
  if (strlen(buf) >= MAX_ADDR_SIZE)
    YIELD(FAIL);

  port = GetInt(PARG(4,1));

  t = GetInt(PARG(4,2));
  timeout.tv_sec  = t / 1000;
  timeout.tv_usec = (t % 1000) * 1000;

  ctx = redisConnectWithTimeout(buf, port, timeout);
  if (ctx->err != 0) {
    fprintf(stderr, "redis_connect: Connection error: '%s'\r\n", ctx->errstr);
    YIELD(FAIL);
  }
  else {
    int r = UnifyInt(PARG(4,3), (BASEINT )ctx);
    if (r != 0) {
      st_insert(ContextTable, (st_data_t )ctx, (st_data_t )0);
    }
    YIELD(r);
  }
}

/* ?-redis_is_disconnected(+CONTEXT). */
static pred
P1_redis_is_disconnected(Frame *Env)
{
  redisContext* ctx;
  ALL_BEGIN(Env);

  ctx = (redisContext* )GetInt(PARG(1, 0));
  if (ctx == 0)
    YIELD(FAIL);

#if 0
  fprintf(stderr, "err: %d, fd: %d, flags: %d\n", ctx->err, ctx->fd, ctx->flags);
  fflush(stderr);
#endif

  if (! st_lookup(ContextTable, (st_data_t )ctx, (st_data_t* )0)) {
    YIELD(FAIL);
  }

  if (ctx->err == REDIS_ERR_EOF)
    YIELD(DET_SUCC);
  else
    YIELD(FAIL);
}

/* ?-redis_free_context(+CONTEXT). */
static pred
P1_redis_free_context(Frame *Env)
{
  int r;
  redisContext* ctx;
  ALL_BEGIN(Env);

  if (SaveReply != 0) {
    freeReplyObject(SaveReply);
    SaveReply = 0;
  }

  ctx = (redisContext* )GetInt(PARG(1,0));
  if (ctx != NULL) {
    if (! st_lookup(ContextTable, (st_data_t )ctx, (st_data_t* )0)) {
      YIELD(FAIL);
    }

    redisFree(ctx);

    r = st_delete(ContextTable, (st_data_t* )&ctx, (st_data_t* )0);
    if (r == 0) {
      YIELD(FAIL);
    }
  }
  else
    YIELD(FAIL);

  YIELD(DET_SUCC);
}


static void clear_command_list(int cn, const char* cs[])
{
  int i;

  for (i = 0; i < cn; i++) {
    if (cs[i] != NULL)
      free((void* )cs[i]);
  }
}

static int prolog_list_to_command(Frame* Env, TERM* t, int max, const char* cs[], size_t cl[])
{
  int r, i, len;
  int n = 0;
  TERM* tb;
  char* p;

  while (IsCons(t)) {
    REALVALUE(t);
    t = BODY(t);
    tb = t;
    REALVALUE(tb);

    if (n >= max) {
      r = -2;
      goto err;
    }

    if (IsAtom(tb)) {
      len = az_get_atom_term_length(Env, tb);
      if (len > 0) {
        p = (char* )malloc((size_t )(len + 1));
        if (p != NULL) {
          if (az_term_to_cstring(Env, tb, p, len + 1) >= 0)
            cs[n++] = p;
        }
      }
    }
    else if (pu_redis_is_prolog_string(tb)) {
      len = az_term_to_cstring_length(Env, tb);
      p = (char* )malloc((size_t )(len + 1));
      if (p != NULL) {
        if (az_term_to_cstring(Env, tb, p, len + 1) >= 0)
          cs[n++] = p;
      }
    }
    else if (! IsUndef(tb)) {
      len = term_to_string(Env, tb, &p);
      if (len < 0) {
        r = -3;
        goto err;
      }
      cs[n++] = p;
    }
    else {
      r = -1;
    err:
      clear_command_list(n, cs);
      return r;
    }

    t++;
  }

  for (i = 0; i < n; i++) {
    cl[i] = strlen(cs[i]);
  }

  return n;
}

static pred
redis_command(Frame *Env, int hate_integer, int conv_type)
{
  redisContext *ctx;
  redisReply *r = NULL;
  char buf[AZ_MAX_ATOM_LENGTH + 1];
  char *command;
  TERM *t1, *t3, *t2;
  BASEINT atype;
  int command_malloced = 0;

  ALL_BEGIN(Env);

  if (SaveReply != 0) {
    freeReplyObject(SaveReply);
    SaveReply = 0;
  }

  ctx = (redisContext* )GetInt(PARG(4,0));

  if (! st_lookup(ContextTable, (st_data_t )ctx, (st_data_t* )0)) {
    YIELD(FAIL);
  }

  command = NULL;

  t1 = PARG(4,1);
  t2 = PARG(4,2);
  REALVALUE(t1);
  if (IS_ATOM(t1)) {
    buf[0] = '\0';
    command = buf;

    if (GetAtom(t1) != ATOM_NIL) { /* ignore ATOM_NIL => "[]" */
      Atom2Asciz(GetAtom(t1), buf);
      if (strlen(buf) >= AZ_MAX_ATOM_LENGTH)
        YIELD(FAIL);

#ifdef DEBUG
      fprintf(stderr, "redis_command: Atom: %lld, %s\r\n", GetAtom(t1), buf);
#endif

    exec_command_simple:
#ifdef DEBUG
      fprintf(stderr, "redis_command: COMMAND: '%s'\r\n", command);
      fflush(stderr);
#endif

      if (strlen(command) == 0) {
#ifdef DEBUG
        fprintf(stderr, "redis_command: command is empty.\r\n");
#endif
        if (command_malloced != 0)
          free(command);

        YIELD(FAIL);
      }

      r = redisCommand(ctx, command);

      if (command_malloced != 0)
        free(command);
    }
  }
  else if (IS_LIST(t1)) {
    if (pu_redis_is_prolog_string(t1)) {
      int len = az_term_to_cstring_length(Env, t1);
      command = (char* )malloc((size_t )(len + 1));
      if (command == 0)
        YIELD(FAIL);

      az_term_to_cstring(Env, t1, command, len + 1);

      command_malloced = 1;
      goto exec_command_simple;
    }
    else {
      int cn;
      const char* cs[MAX_COMMAND_ARGS_NUM];
      size_t cl[MAX_COMMAND_ARGS_NUM];

      cn = prolog_list_to_command(Env, t1, MAX_COMMAND_ARGS_NUM, cs, cl);
      if (cn <= 0)
        YIELD(FAIL);
      else {
#if 0
        int i;
        for (i = 0; i < cn; i++) {
          fprintf(stderr, "[%s]\r\n", cs[i]);
        }
#endif
        r = redisCommandArgv(ctx, cn, cs, cl);
        clear_command_list(cn, cs);
      }
    }
  }
  else {
    AZ_ERROR(9); /* Illegal Argument */
  }

  if (r == (void* )0) { /* ex. client timeout */
    YIELD(FAIL);
  }

  atype = type_to_atom(r->type);

  t3 = PARG(4,3);
  REALVALUE(t3);
  switch (r->type) {
  case REDIS_REPLY_ARRAY:
    {
      int ret;

      ReplyArrayStart = 0;
      ret = make_list_from_reply_array(Env, t3, r, &ReplyArrayStart,
                                       MaxReplyNum, hate_integer, conv_type);
      if (ret < 0) {
        goto fail;
      }
      else if (ret > 0) {
        SaveReply = r;
        SaveHateInteger = hate_integer;
        SaveConvType    = conv_type;
      }
    }
    break;

  case REDIS_REPLY_NIL:
    if (UnifyAtom(t3, ATOM_NIL) == 0)
      goto fail;
    break;

  case REDIS_REPLY_INTEGER:
    if (hate_integer != 0) {
      atype = type_to_atom(REDIS_REPLY_STRING);
      sprintf(buf, "%lld", r->integer);
      if (unify_term_or_list(Env, t3, buf, conv_type) == 0)
        goto fail;
    }
    else {
      if (UnifyInt(t3, (BASEINT )r->integer) == 0)
        goto fail;
    }
    break;

  case REDIS_REPLY_STATUS:
  case REDIS_REPLY_ERROR:
  case REDIS_REPLY_STRING:
    if (r->len == 0) {
      if (UnifyAtom(t3, ATOM_NIL) == 0)
        goto fail;
    }
    else {
      int v;

      if (conv_type == CONV_ATOM) {
        v = make_one_element_list(Env, t3, r->str);
        if (v != 0)
          goto fail;
      }
      else {
        v = unify_term_or_list(Env, t3, r->str, conv_type);
        //fprintf(stderr, "unify_term_or_list: %d\r\n", v);
        if (v == 0)
          goto fail;
      }
    }
    break;

  default:
    goto fail;
    break;
  }

  if (unify_atom(t2, atype) == 0) {
    //fprintf(stderr, "unify_atom: fail\r\n");
  fail:
    freeReplyObject(r);
    YIELD(FAIL);
  }

  if (SaveReply != r)
    freeReplyObject(r);

  YIELD(DET_SUCC);
}

/* ?-redis_command(+CONTEXT, +COMMAND, -RETURN_TYPE, -RETURN_VALUE). */
static pred
P4_redis_command(Frame *Env)
{
  return redis_command(Env, 0, 0);
}

/* ?-redis_command_to_term(+CONTEXT, +COMMAND, -RETURN_TYPE, -RETURN_VALUE). */
static pred
P4_redis_command_to_term(Frame *Env)
{
  return redis_command(Env, 0, CONV_TERM);
}

/* ?-redis_command_to_atom_list(+CONTEXT, +COMMAND, -RETURN_TYPE, -RETURN_VALUE). */
static pred
P4_redis_command_to_atom_list(Frame *Env)
{
  return redis_command(Env, 0, CONV_ATOM);
}

/* ?-redis_command_hate_integer(+CONTEXT, +COMMAND, -RETURN_TYPE, -RETURN_VALUE). */
static pred
P4_redis_command_hate_integer(Frame *Env)
{
  return redis_command(Env, 1, 0);
}

/* ?-redis_next_val(-VAL, -REMAIN_NUM). */
static pred
P2_redis_next_val(Frame *Env)
{
  int remain;
  int r;

  ALL_BEGIN(Env);

  if (SaveReply == 0) {
    if (UnifyAtom(PARG(2, 0), ATOM_NIL) == 0)
      YIELD(FAIL);

    YIELD(unify_int(PARG(2, 1), (SBASEINT )0));
  }

  r = make_list_from_reply_array(Env, PARG(2, 0), SaveReply, &ReplyArrayStart,
                                 MaxReplyNum, SaveHateInteger, SaveConvType);
  if (r < 0) {
#ifdef DEBUG
    fprintf(stderr, "P2_redis_next_val(): make_list_from_reply_array() fail.\n");
#endif
    freeReplyObject(SaveReply);
    SaveReply = 0;
    YIELD(FAIL);
  }

  remain = r;

  if (r == 0) {
    freeReplyObject(SaveReply);
    SaveReply = 0;
  }

  YIELD(unify_int(PARG(2, 1), (SBASEINT )remain));
}

/* ?-redis_set_max_reply_num(+NUM). */
static pred
P1_redis_set_max_reply_num(Frame *Env)
{
  int n;

  ALL_BEGIN(Env);

  n = (int )GetInt(PARG(1, 0));
  if (n <= 0)
    YIELD(FAIL);

  MaxReplyNum = n;
  YIELD(DET_SUCC);
}

#if 0
/* ?-redis_set_reply_value_atom(+on/off). */
static pred
P1_redis_set_reply_value_atom(Frame *Env)
{
  BASEINT a;

  ALL_BEGIN(Env);

  a = GetAtom(PARG(1, 0));
  if (a == ON_ATOM)
    UnifyAtomMode = 1;
  else if (a == OFF_ATOM)
    UnifyAtomMode = 0;
  else
    INVALID_ARG;

  YIELD(DET_SUCC);
}
#endif

/* ?-redis_version(-VERSION). */
static pred
P1_redis_version(Frame *Env)
{
  ALL_BEGIN(Env);

  if (UnifyAtom(PARG(1,0), REDIS_VERSION_ATOM) == 0)
    YIELD(FAIL);

  YIELD(DET_SUCC);
}

/* ?-redis_string_replace(+SOURCE, +FROM, +TO, -RESULT). */
static pred
P4_redis_string_replace(Frame *Env)
{
  char* source;
  char* from;
  char* to;
  char* result;
  BASEINT a;
  int r;
  int n;

  ALL_BEGIN(Env);

  r = get_term_str(Env, PARG(4, 0), &source);
  if (r < 0) YIELD(FAIL);
  r = get_term_str(Env, PARG(4, 1), &from);
  if (r < 0) {
    free(source);
    YIELD(FAIL);
  }
  r = get_term_str(Env, PARG(4, 2), &to);
  if (r < 0) {
    free(source);
    free(from);
    YIELD(FAIL);
  }

  n = pu_string_replace(source, from, to, &result);
  //fprintf(stderr, "result: [%s]\n", result);

  free(from);
  free(to);

  if (n < 0) {
    free(source);
    YIELD(FAIL);
  }

  a = PutAtom(Env, result);

  free(source);
  if (n > 0) free(result);
  YIELD(UnifyAtom(PARG(4, 3), (SBASEINT )a));
}
