void SingleKeyspaceDB::OnExpiryTimer()
{
	uint64_t	expiryTime;
	Cursor		cursor;
	ByteString	key;

	Log_Trace();
	
	table->Iterate(NULL, cursor);	
	kdata.Set("!!t:");
	if (!cursor.Start(kdata))
		ASSERT_FAIL();
	cursor.Close();
	
	if (kdata.length < 2)
		ASSERT_FAIL();
	
	if (kdata.buffer[0] != '!' || kdata.buffer[1] != '!')
		ASSERT_FAIL();

	ReadExpiryTime(kdata, expiryTime, key);
	table->Delete(NULL, kdata);
	table->Delete(NULL, key);

	WriteExpiryKey(kdata, key);
	table->Delete(NULL, kdata);
	
	Log_Trace("Expiring key: %.*s", key.length, key.buffer);

	InitExpiryTimer();
}

static void
test_wrong_op_fail(void)
{
	ASSERT_FAIL("'a' =! 'a'", PE_INVALID_EXPRESSION);
	ASSERT_FAIL("'a' => 'a'", PE_INVALID_EXPRESSION);
	ASSERT_FAIL("'a' =< 'a'", PE_INVALID_EXPRESSION);
}

/*this function "links" IoTHub to the serialization library*/
static IOTHUBMESSAGE_DISPOSITION_RESULT IoTHubMessage(IOTHUB_MESSAGE_HANDLE message, void* userContextCallback)
{
    const unsigned char* buffer;
    size_t size;
    if (IoTHubMessage_GetByteArray(message, &buffer, &size) != IOTHUB_MESSAGE_OK)
    {
        ASSERT_FAIL("unable to IoTHubMessage_GetByteArray");
    }
    else
    {
        /*buffer is not zero terminated*/
        char* buffer_string = (char*)malloc(size+1);
        ASSERT_IS_NOT_NULL(buffer_string);

        if (memcpy(buffer_string, buffer, size) == 0)
        {
            ASSERT_FAIL("memcpy failed for buffer");
        }
        else
        {
            buffer_string[size] = '\0';
            EXECUTE_COMMAND(userContextCallback, buffer_string);
        }
        free(buffer_string);
    }
    return IOTHUBMESSAGE_ACCEPTED;
}

void ConfigQuorumContext::OnMessage(ReadBuffer buffer)
{
    char proto;
    
    Log_Trace("%R", &buffer);

    if (buffer.GetLength() < 2)
        ASSERT_FAIL();

    proto = buffer.GetCharAt(0);
    ASSERT(buffer.GetCharAt(1) == ':');
    
    switch(proto)
    {
        case PAXOSLEASE_PROTOCOL_ID:        // 'L':
            OnPaxosLeaseMessage(buffer);
            break;
        case PAXOS_PROTOCOL_ID:             // 'P':
            OnPaxosMessage(buffer);
            break;
        case CATCHUP_PROTOCOL_ID:           // 'C'
            OnCatchupMessage(buffer);
            break;
        default:
            ASSERT_FAIL();
            break;
    }
}

CTEST2(reorder, shuffle)
{
  memcpy(data->buffer, data->fororder, data->N * sizeof(*data->buffer));

  /* do this 10 times */
  for(idx_t e=0; e < 10; ++e) {
    shuffle_idx(data->buffer, data->N);
    idx_t same = 0;
    for(idx_t x=0; x < data->N; ++x) {
      if(data->fororder[x] == data->buffer[x]) {
        ++same;
      }
    }

    /* arbitrary */
    if(same > 4) {
      ASSERT_FAIL();
    }

    /* check for duplicate/missing values */
    memcpy(data->fororder, data->buffer, data->N * sizeof(*data->buffer));
    quicksort(data->fororder, data->N);
    for(idx_t x=0; x < data->N; ++x) {
      if(data->fororder[x] != x) {
        ASSERT_FAIL();
      }
    }
  }
}

void ContextTransport::OnMessage(uint64_t nodeID, ReadBuffer msg)
{
    int         nread;
    char        proto;
    
    Log_Trace("%R", &msg);
    
    if (msg.GetLength() < 2)
        ASSERT_FAIL();

    nread = msg.Readf("%c:", &proto);
    if (nread < 2)
        ASSERT_FAIL();

    msg.Advance(2);
    
    switch (proto)
    {
        case PROTOCOL_CLUSTER:
            OnClusterMessage(nodeID, msg);
            break;
        case PROTOCOL_QUORUM:
            OnQuorumMessage(nodeID, msg);
            break;
        default:
            ASSERT_FAIL();
            break;
    }
}

static void
test_spaces_in_op_fail(void)
{
	ASSERT_FAIL("'a' ! = 'a'", PE_INVALID_EXPRESSION);
	ASSERT_FAIL("'a' = = 'a'", PE_INVALID_EXPRESSION);
	ASSERT_FAIL("'a' > = 'a'", PE_INVALID_EXPRESSION);
	ASSERT_FAIL("'a' < = 'a'", PE_INVALID_EXPRESSION);
}

static void
test_spaces_and_fail_position_correct(void)
{
	ASSERT_FAIL("  'b' c", PE_INVALID_EXPRESSION)
	assert_string_equal("'b' c", get_last_position());

	ASSERT_FAIL("  a b", PE_INVALID_EXPRESSION)
	assert_string_equal("a b", get_last_position());
}

static int IoTHubCallback(void* context, const char* data, size_t size)
{
    EXPECTED_SEND_DATA* expectedData = (EXPECTED_SEND_DATA*)context;
    int result = 0; // 0 means "keep processing"

    if (expectedData != NULL)
    {
        if (Lock(expectedData->lock) != LOCK_OK)
        {
            ASSERT_FAIL("unable to lock");
        }
        else
        {
            if (
                (strlen(expectedData->expectedString) == size) &&
                (memcmp(expectedData->expectedString, data, size) == 0)
                )
            {
                expectedData->wasFound = true;
                result = 1;
            }
            (void)Unlock(expectedData->lock);
        }
    }
    return result;
}

static bool screen_key_event_handler(void *usr, key_event_t *e)
{
    PyObject *func = (PyObject *)usr;
    PyObject *args = PyTuple_New(1);
    PyTuple_SetItem(args, 0, PyInt_FromLong(e->key_code));
    PyObject *result = PyObject_CallObject(func, args);

    bool ret = false;
    if(!PyBool_Check(result)) {
        ERROR("python key handler did not return a boolean\n");
        ret = false;
    } else if(result == Py_True) {
        ret = true;
    } else if(result == Py_False) {
        ret = false;
    } else {
        ASSERT_FAIL("screen_key_event_handler: this should never happen...");
    }

    // some cleanup
    Py_DECREF(result);
    Py_DECREF(args);

    return ret;
}

const Endpoint ReplicatedConfig::GetEndpoint(unsigned i)
{
	if (i < 0 || i >= numNodes)
		ASSERT_FAIL();
	
	return endpoints[i];
}

void ReplicatedKeyspaceDB::FailKeyspaceOps()
{
	Log_Trace();

	KeyspaceOp	**it;
	KeyspaceOp	*op;
	for (it = ops.Head(); it != NULL; /* advanded in body */)
	{
		op = *it;
		
		it = ops.Remove(it);
		op->status = false;
		if (op->service)
			op->service->OnComplete(op);
		else
		{
			assert(op->type == KeyspaceOp::EXPIRE);
			delete op;
		}
	}

	expiryAdded = false;

	if (ops.Length() > 0)
		ASSERT_FAIL();
}

void RGB24Buffer::drawFlowBuffer(FlowBuffer *src, int32_t y, int32_t x)
{
    int i,j;
    for (i = 0; i < src->h; i++)
    {
        for (j = 0; j < src->w; j++)
        {
            if (!src->isElementKnown(i,j))
            {
                continue;
            }
            FlowElement vec = src->element(i,j);

#ifdef ASSERTS
            Vector2d32 res(vec);
            res += Vector2d32(j, i);
            if (!this->isValidCoord(res))
            {
                ASSERT_FAIL("Overflow in the flow");
            }
#endif

            RGBColor color = RGBColor(128 + (vec.x() * 5),128 + (vec.y() * 5), 0);

            this->element(i + y, j + x) = color;
        }
    }
}

char ReadBuffer::GetCharAt(unsigned i) const
{
    if (i > length - 1)
        ASSERT_FAIL();
    
    return *(buffer + i);
}

bool IOProcessor::Add(IOOperation* ioop)
{
	if (ioop->active)
	{
		ASSERT_FAIL();
//		return true;
	}

	// empty buffer indicates that we are waiting for accept event
	if (ioop->type == TCP_READ && ioop->data.buffer == NULL)
		return StartAsyncAccept(ioop);

	// zero length indicates that we are waiting for connect event
	if (ioop->type == TCP_WRITE && ioop->data.length == 0)
		return StartAsyncConnect(ioop);

	switch (ioop->type)
	{
	case TCP_READ:
	case UDP_READ:
		return RequestReadNotification(ioop);
		break;
	case TCP_WRITE:
	case UDP_WRITE:
		return RequestWriteNotification(ioop);
		break;
	}

	return false;
}

 static int IoTHubCallback(void* context, const char* data, size_t size)
 {
     int result = 0; // 0 means "keep processing"
     EXPECTED_SEND_DATA* expectedData = (EXPECTED_SEND_DATA*)context;
     //printf("Received: %*.*s\n", (int)size, (int)size, data);
     if (expectedData != NULL)
     {
         if (Lock(expectedData->lock) != LOCK_OK)
         {
             ASSERT_FAIL("unable to lock");
         }
         else
         {
             if (size != strlen(expectedData->expectedString))
             {
                 result = 0;
             }
             else
             {
                 if (memcmp(expectedData->expectedString, data, size) == 0)
                 {
                     expectedData->wasFound = true;
                     result = 1;
                 }
                 else
                 {
                     result = 0;
                 }
             }
             (void)Unlock(expectedData->lock);
         }
     }
     return result;
 }

static EXPECTED_SEND_DATA* EventData_Create(void)
{
    EXPECTED_SEND_DATA* result = (EXPECTED_SEND_DATA*)malloc(sizeof(EXPECTED_SEND_DATA));
    if (result != NULL)
    {
        if ((result->lock = Lock_Init()) == NULL)
        {
            ASSERT_FAIL("unable to Lock_Init");
        }
        else
        {
            char temp[1000];
            char* tempString;
            time_t t = time(NULL);
            sprintf(temp, TEST_EVENT_DATA_FMT, ctime(&t), g_iotHubTestId);
            if ((tempString = (char*)malloc(strlen(temp) + 1)) == NULL)
            {
                Lock_Deinit(result->lock);
                free(result);
                result = NULL;
            }
            else
            {
                strcpy(tempString, temp);
                result->expectedString = tempString;
                result->wasFound = false;
                result->dataWasRecv = false;
            }
        }
    }
    return result;
}

static void sanity_checks(unsigned int check)
{
	if (check >= NUM_TEST_TYPES) {
		CTEST_LOG("WARNING: argument check out of bounds");
		ASSERT_FAIL();
	}
}

void ReplicatedLog::OnMessage(PaxosMessage& imsg)
{
    Log_Trace();
    bool processed;

    processed = false;
    if (imsg.type == PAXOS_PREPARE_REQUEST)
        processed = OnPrepareRequest(imsg);
    else if (imsg.IsPrepareResponse())
        processed = OnPrepareResponse(imsg);
    else if (imsg.type == PAXOS_PROPOSE_REQUEST)
        processed = OnProposeRequest(imsg);
    else if (imsg.IsProposeResponse())
        processed = OnProposeResponse(imsg);
    else if (imsg.IsLearn())
        processed = OnLearnChosen(imsg);
    else if (imsg.type == PAXOS_REQUEST_CHOSEN)
        processed = OnRequestChosen(imsg);
    else if (imsg.type == PAXOS_START_CATCHUP)
        processed = OnStartCatchup(imsg);
    else
        ASSERT_FAIL();

    if (processed)
        context->OnMessageProcessed();
}

int base_test()
{
	int64_t llval;
	fc_conf_t conf;
	fc_item_t item, item2;

	fc_conf_init(&conf);
	conf.ksem = SEMKEY;
	conf.kmem = SHMKEY;
	conf.memsize = SHMSIZE;

	memset(&item, 0, sizeof(item));
	memset(&item2, 0, sizeof(item2));

	ASSERT_OK(fc_init(&conf)==0,"[%d:%s]",ERRPAIR);
	SETKV(item2,"yingyuan","");
	ASSERT_FAIL(fc_get(&item2)==0,"[%d:%s]%s:%s",ERRPAIR,ITPAIR(item2));
	SETKV(item,"yingyuan","Cheng Yingyuan");
	ASSERT_OK(fc_add(&item)==0,"[%d:%s]%s:%s",ERRPAIR,ITPAIR(item));
	ASSERT_OK(fc_get(&item2)==0,"[%d:%s]%s:%s",ERRPAIR,ITPAIR(item2));
	ASSERT_EQ(item,item2);
	ASSERT_FAIL(fc_add(&item)==0,"[%d:%s]%s:%s",ERRPAIR,ITPAIR(item));
	//SETKV(item2, "yingyuanx", "xxxxxx");
	//ASSERT_OK(fc_get(&item2)==0,"[%d:%s]%s:%s",ERRPAIR,ITPAIR(item2));
	//ASSERT_EQ(item,item2);
	//SETKV(item, "yingyuan", "REAL WORLD");
	//ASSERT_OK(fc_get(&item2)==0,"[%d:%s]%s:%s",ERRPAIR,ITPAIR(item2));
	//ASSERT_EQ(item,item2);
	//SETKV(item,"yingyuanx","Cheng");
	//SETKV(item2,"yingyuanx","");
	//ASSERT_OK(fc_get(&item2)==0,"[%d:%s]%s:%s",ERRPAIR,ITPAIR(item2));
	//ASSERT_EQ(item,item2);
	SETKV(item2,"age","");
	item2.value = (char *)&llval;
	item2.vlen = sizeof(llval);
	llval=1000;
	ASSERT_OK(llval==2000,"inlegal llval %llu",llval);
	//SETKV(item2, "yingyuanx","");
	/*fc_debug(stdout, 1);
	fc_dump(stdout);*/
	SETKV(item,"yingyuan","");
	//ASSERT_FAIL(fc_get(&item)==0,"%s","error test fail");
	/*ASSERT_OK(fc_destroy()==0,"%s","base test");*/
	printf("PASS!!!\n");
	return 0;
}