File content as of revision 13:61e0cc093180:

#include "Milkcocoa.h"
#include "OldTimer.h"

#if 0
extern RawSerial pc;

#define DBG(x)  x
#else
#define DBG(x)
#endif

DataElement::DataElement() {
	json_msg[0] = '\0';
	strcpy(json_msg,"{\"params\":{");
}

DataElement::DataElement(char *json_string) {
	json_msg[0] = '\0';
	strcpy(json_msg,json_string);
}

void DataElement::setValue(const char *key, const char *v) {
	char json_string[64];
	if( json_msg[strlen(json_msg)-1] != '{' )
	{
		strcat(json_msg,",");
	}
	sprintf(json_string,"\"%s\":\"%s\"",key,v);
	strcat(json_msg,json_string);
}

void DataElement::setValue(const char *key, int v) {
	char json_string[64];
	if( json_msg[strlen(json_msg)-1] != '{' )
	{
		strcat(json_msg,",");
	}
	sprintf(json_string,"\"%s\":\"%d\"",key,v);
	strcat(json_msg,json_string);
}

void DataElement::setValue(const char *key, double v) {
	char json_string[64];
	if( json_msg[strlen(json_msg)-1] != '{' )
	{
		strcat(json_msg,",");
	}
	sprintf(json_string,"\"%s\":\"%f\"",key,v);
	strcat(json_msg,json_string);
}

char *DataElement::getString(const char *key) {
    static char _word[64];
    char *p;
	int i=0;
	
	strcpy(_word , "\"\0");
    strcat(_word , key );
    strcat(_word , "\"" );
 
    p = strstr( (char*)json_msg , _word ) + 2 + strlen(_word);
	
    while( (p[i] != ',')&&(p[i] != '\n')&&(p[i] != '\"') )
    {
        _word[i] = p[i];
        i++;
    }
    _word[i] = '\0';
	
	return _word;
}

int DataElement::getInt(const char *key) {
	return atoi(getString(key));
}

float DataElement::getFloat(const char *key) {
	return atof(getString(key));
}

char *DataElement::toCharArray() {
	if( json_msg[strlen(json_msg)-1] != '{' )
	{
		strcat(json_msg,"}");
	}
	strcat(json_msg,"}");
	
	return(json_msg);
}

Milkcocoa::Milkcocoa(NetworkInterface* nif){
	ipstack = new MQTTInterface(nif);
	client = new MClient(ipstack);
	
	for (int i=0; i<MILKCOCOA_SUBSCRIBERS; i++) {
		milkcocoaSubscribers[i] = NULL;
	}
}

Milkcocoa::Milkcocoa(NetworkInterface* nif, const char *host, uint16_t port, const char *_app_id, const char *client_id){
	ipstack = new MQTTInterface(nif);
	client = new MClient(ipstack);
	strcpy(servername,host);
	portnum = port;
	strcpy(app_id,_app_id);
	strcpy(_clientid,client_id);
	strcpy(username,"sdammy");
	strcpy(password,app_id);
	
	for (int i=0; i<MILKCOCOA_SUBSCRIBERS; i++) {
		milkcocoaSubscribers[i] = NULL;
	}
	
#ifdef __MILKCOCOA_THREAD
	setLoopCycle(5000);
	
	cycleThread1.start(Milkcocoa::threadStarter1,this);
	cycleThread2.start(Milkcocoa::threadStarter2,this);
#endif
}

Milkcocoa::Milkcocoa(NetworkInterface* nif, const char *host, uint16_t port, const char *_app_id, const char *client_id, char *_session){
	ipstack = new MQTTInterface(nif);
	client = new MClient(ipstack);
	strcpy(servername,host);
	portnum = port;
	strcpy(app_id,_app_id);
	strcpy(_clientid,client_id);
	strcpy(username,_session);
	strcpy(password,app_id);
	
	for (int i=0; i<MILKCOCOA_SUBSCRIBERS; i++) {
		milkcocoaSubscribers[i] = NULL;
	}
	
#ifdef __MILKCOCOA_THREAD
	setLoopCycle(5000);
	
	cycleThread1.start(Milkcocoa::threadStarter1,this);
	cycleThread2.start(Milkcocoa::threadStarter2,this);
#endif
}

Milkcocoa* Milkcocoa::createWithApiKey(NetworkInterface* nif, const char *host, uint16_t port, const char *_app_id, const char *client_id, char *key, char *secret) {
	char session[60];
	sprintf(session, "k%s:%s", key, secret);
	return new Milkcocoa(nif, host, port, _app_id, client_id, session);
}

void Milkcocoa::set_network_settinng(const char *host, uint16_t port, const char *_app_id, const char *client_id){
	strcpy(servername,host);
	portnum = port;
	strcpy(app_id,_app_id);
	strcpy(_clientid,client_id);
	strcpy(username,"sdammy");
	strcpy(password,app_id);
}

void Milkcocoa::connect() {

	if(client->isConnected())
		return;

	if(client->connect(servername, portnum)!=0) {
		DBG(pc.printf("Network connect err\r\n");)
		return;
	}

	MQTTPacket_connectData data = MQTTPacket_connectData_initializer;
	data.keepAliveInterval = 20;
	data.cleansession = 1;
	data.MQTTVersion = 4;
	data.clientID.cstring = _clientid;
	data.username.cstring = username;
	data.password.cstring = password;

	if (client->connect(data) != 0)  {
		DBG(pc.printf("Milkcocoa connect err\r\n");)
		return;
	}

}

bool Milkcocoa::push(const char *path, DataElement dataelement) {
	milkcocoa_message_t *message = message_box.alloc();
	char *buf;
	
	if(message == NULL)	return false;

	sprintf(message->topic, "%s/%s/push", app_id, path);
	buf = dataelement.toCharArray();
	strcpy(message->message , buf);
	
	osStatus stat = message_box.put(message);
	
	if( stat != osOK ) return false;

	return true;
}

bool Milkcocoa::push(const char *path, char *data) {

	milkcocoa_message_t *message = message_box.alloc();
	char *buf;
	
	if(message == NULL)	return false;

	sprintf(message->topic, "%s/%s/push", app_id, path);
	
	strcpy(message->message , "{\"params\":");
	strcat(message->message , data);
	strcat(message->message , "}");
	
	osStatus stat = message_box.put(message);
	
	if( stat != osOK ) return false;

	return true;
}

bool Milkcocoa::send(const char *path, DataElement dataelement) {
	milkcocoa_message_t *message = message_box.alloc();
	char *buf;
	
	if(message == NULL)	return false;

	sprintf(message->topic, "%s/%s/send", app_id, path);
	buf = dataelement.toCharArray();
	strcpy(message->message , buf);
	
	osStatus stat = message_box.put(message);
	
	if( stat != osOK ) return false;

	return true;
}

bool Milkcocoa::send(const char *path, char *data) {
	milkcocoa_message_t *message = message_box.alloc();
	char *buf;
	
	if(message == NULL)	return false;

	sprintf(message->topic, "%s/%s/send", app_id, path);
	
	strcpy(message->message , "{\"params\":");
	strcat(message->message , data);
	strcat(message->message , "}");
	
	osStatus stat = message_box.put(message);
	
	if( stat != osOK ) return false;

	return true;
}

void Milkcocoa::loop() {
	connect();
#ifndef __MILKCOCOA_THREAD
	send_massage();
#endif
	client->yield(1);
}

void Milkcocoa::send_massage(){
	while(1){
		osEvent evt = message_box.get(0);
		if (evt.status == osEventMail) {
			milkcocoa_message_t *message = (milkcocoa_message_t*)evt.value.p;
			MQTT::Message mq_message;
			if(message != NULL) {
				mq_message.qos = MQTT::QOS0;
				mq_message.retained = 0;
				mq_message.dup = false;
				mq_message.payload = (void*)message->message;
				mq_message.payloadlen = strlen(message->message);
				printf("milkcocoa send\r\n");
				client->publish(message->topic, mq_message);
				
				message_box.free(message);
			}
		}
		else{
			break;
		}
	}
}

bool Milkcocoa::on(const char *path, const char *event, GeneralFunction cb) {
	MilkcocoaSubscriber *sub = new MilkcocoaSubscriber(cb);
	sprintf(sub->topic, "%s/%s/%s", app_id, path, event);

	if (client->subscribe(sub->topic, MQTT::QOS0, cb) != 0)  {
		DBG(pc.printf("Milkcocoa subscribe err\r\n");)
		return false;
	}
	for (int i=0; i<MILKCOCOA_SUBSCRIBERS; i++) {
		if (milkcocoaSubscribers[i] == sub) {
			return false;
		}
	}
	for (int i=0; i<MILKCOCOA_SUBSCRIBERS; i++) {
		if (milkcocoaSubscribers[i] == 0) {
			milkcocoaSubscribers[i] = sub;
			return true;
		}
	}
	return true;
}

#ifdef __MILKCOCOA_THREAD
void Milkcocoa::setLoopCycle(int cycle) {
	loop_cycle = cycle;
}
void Milkcocoa::start() {
	cycleThread1.signal_set(START_THREAD);
	cycleThread2.signal_set(START_THREAD);
}

void Milkcocoa::cycle_Thread1(void) {
	cycleThread1.signal_wait(START_THREAD);
	while(1) {
    	OldTimer timer;
		timer.start();
		connect();
				
		client->yield(RECV_TIMEOUT);
		
		int sub_time = loop_cycle - timer.read();
		if( sub_time > 0 ){
			 Thread::wait(sub_time);
		}
		timer.stop();
	}
}
void Milkcocoa::cycle_Thread2(void) {
	cycleThread2.signal_wait(START_THREAD);
	while(1) {

        osEvent evt = message_box.get();
        if (evt.status == osEventMail) {
        	milkcocoa_message_t *message = (milkcocoa_message_t*)evt.value.p;
        	MQTT::Message mq_message;
        	
        	if(message == NULL)	break;
        	
			mq_message.qos = MQTT::QOS0;
			mq_message.retained = 0;
			mq_message.dup = false;
			mq_message.payload = (void*)message->message;
			mq_message.payloadlen = strlen(message->message);
		
			client->publish(message->topic, mq_message);
			
			message_box.free(message);
		}
		Thread::wait(1000);
	}
}

void Milkcocoa::threadStarter1(void const *p) {
  Milkcocoa *instance = (Milkcocoa*)p;
  instance->cycle_Thread1();
}
void Milkcocoa::threadStarter2(void const *p) {
  Milkcocoa *instance = (Milkcocoa*)p;
  instance->cycle_Thread2();
}
#endif

MilkcocoaSubscriber::MilkcocoaSubscriber(GeneralFunction _cb) {
  cb = _cb;
}