c-libp2p/yamux/yamux.c

#include <string.h>
#include <unistd.h>
#include "varint.h"
#include "libp2p/yamux/session.h"
#include "libp2p/yamux/yamux.h"
#include "libp2p/net/protocol.h"
#include "libp2p/net/stream.h"
#include "libp2p/conn/session.h"
#include "libp2p/utils/logger.h"

/**
 * Determines if this protocol can handle the incoming message
 * @param incoming the incoming data
 * @param incoming_size the size of the incoming data buffer
 * @returns true(1) if it can handle this message, false(0) if not
 */
int yamux_can_handle(const struct StreamMessage* msg) {
	char *protocol = "/yamux/1.0.0\n";
	int protocol_size = strlen(protocol);
	// is there a varint in front?
	size_t num_bytes = 0;
	if (msg->data[0] != protocol[0] && msg->data[1] != protocol[1]) {
		varint_decode(msg->data, msg->data_size, &num_bytes);
	}
	if (msg->data_size >= protocol_size - num_bytes) {
		if (strncmp(protocol, (char*) &msg->data[num_bytes], protocol_size) == 0)
			return 1;
	}
	return 0;
}

/**
 * the yamux stream received some bytes. Process them
 * @param stream the stream that the data came in on
 * @param msg the message
 * @param incoming the stream buffer
 */
void yamux_read_stream(struct yamux_stream* stream, struct StreamMessage* msg) {
	struct Libp2pVector* handlers = stream->userdata;
	int retVal = libp2p_protocol_marshal(msg, stream->session->session_context, handlers);
	if (retVal == -1) {
		// TODO handle error condition
		libp2p_logger_error("yamux", "Marshalling returned error.\n");
	} else if (retVal > 0) {
		// TODO handle everything went okay
		libp2p_logger_debug("yamux", "Marshalling was successful. We should continue processing.\n");
	} else {
		// TODO we've been told we shouldn't do anything anymore
		libp2p_logger_debug("yamux", "Marshalling was successful. We should stop processing.\n");
	}
	return;
}

/***
 * Send the yamux protocol out the default stream
 * NOTE: if we initiate the connection, we should expect the same back
 * @param context the SessionContext
 * @returns true(1) on success, false(0) otherwise
 */
int yamux_send_protocol(struct SessionContext* context) {
	char* protocol = "/yamux/1.0.0\n";
	struct StreamMessage outgoing;
	outgoing.data = (uint8_t*)protocol;
	outgoing.data_size = strlen(protocol);
	if (!context->default_stream->write(context, &outgoing))
		return 0;
	return 1;
}

/***
 * Check to see if the reply is the yamux protocol header we expect
 * NOTE: if we initiate the connection, we should expect the same back
 * @param context the SessionContext
 * @returns true(1) on success, false(0) otherwise
 */
int yamux_receive_protocol(struct SessionContext* context) {
	char* protocol = "/yamux/1.0.0\n";
	struct StreamMessage* results = NULL;
	int retVal = 0;

	if (!context->default_stream->read(context, &results, 30)) {
		libp2p_logger_error("yamux", "receive_protocol: Unable to read results.\n");
		goto exit;
	}
	// the first byte is the size, so skip it
	char* ptr = strstr((char*)&results->data[1], protocol);
	if (ptr == NULL || ptr - (char*)results->data > 1) {
		goto exit;
	}
	retVal = 1;
	exit:
	libp2p_stream_message_free(results);
	return retVal;
}

/***
 * Handles the message
 * @param msg the incoming message
 * @param incoming_size the size of the incoming data buffer
 * @param session_context the information about the incoming connection
 * @param protocol_context the protocol-dependent context
 * @returns 0 if the caller should not continue looping, <0 on error, >0 on success
 */
int yamux_handle_message(const struct StreamMessage* msg, struct SessionContext* session_context, void* protocol_context) {
	// they've asked to swicth to yamux. Do the switch and return 0 so that nothing else listens on this stream
	struct yamux_session* yamux = yamux_session_new(NULL, session_context, yamux_session_server, protocol_context);
	uint8_t* buf = (uint8_t*) malloc(msg->data_size);
	if (buf == NULL)
		return -1;
	memcpy(buf, msg->data, msg->data_size);
	for(;;) {
		int retVal = yamux_decode(yamux, msg->data, msg->data_size);
		free(buf);
		buf = NULL;
		if (!retVal)
			break;
		else { // try to read more from this stream
			// TODO need more information as to what this loop should do
		}
	}

	/*
	struct Libp2pVector* handlers = (struct Libp2pVector*)protocol_context;
	uint8_t* results = NULL;
	size_t bytes_read = 0;
	int numRetries = 0;
	int retVal = 0;
	int max_retries = 100; // try for 5 minutes
	for(;;) {
   		// try to read for 5 seconds
   	    if (session_context->default_stream->read(session_context, &results, &bytes_read, 5)) {
   	    	// we read something from the network. Process it.
   	    	// NOTE: If it is a multistream protocol that we are receiving, ignore it.
   	    	if (yamux_can_handle(results, bytes_read))
   	    		continue;
   	    	numRetries = 0;
   	   	retVal = libp2p_protocol_marshal(results, bytes_read, session_context, handlers);
   	   	if (results != NULL)
   	   		free(results);
   	   	// exit the loop on error (or if they ask us to no longer loop by returning 0)
   	   	if (retVal <= 0)
   	   		break;
   	    } else {
   	    		// we were unable to read from the network.
   	   	    // if it timed out, we should try again (if we're not out of retries)
   	    		if (numRetries >= max_retries)
   	    			break;
   	    		numRetries++;
   	    }
	}
	*/
	return 0;
}

/**
 * Shutting down. Clean up any memory allocations
 * @param protocol_context the context
 * @returns true(1)
 */
int yamux_shutdown(void* protocol_context) {
	return 0;
}

struct Libp2pProtocolHandler* yamux_build_protocol_handler(struct Libp2pVector* handlers) {
	struct Libp2pProtocolHandler* handler = libp2p_protocol_handler_new();
	if (handler != NULL) {
		handler->context = handlers;
		handler->CanHandle = yamux_can_handle;
		handler->HandleMessage = yamux_handle_message;
		handler->Shutdown = yamux_shutdown;
	}
	return handler;
}

int libp2p_yamux_close(void* stream_context) {
	//TODO: Implement
	return 0;
}

/**
 * Read from the network, expecting a yamux frame.
 * NOTE: This will also dispatch the frame to the correct protocol
 * @param stream_context the YamuxContext
 * @param message the resultant message
 * @param timeout_secs when to give up
 * @returns true(1) on success, false(0) on failure
 */
int libp2p_yamux_read(void* stream_context, struct StreamMessage** message, int timeout_secs) {
	struct YamuxContext* ctx = (struct YamuxContext*)stream_context;
	struct Stream* parent_stream = ctx->stream->parent_stream;

	struct StreamMessage* incoming;
	if (!parent_stream->read(parent_stream->stream_context, &incoming, timeout_secs))
		return 0;

	// we've got bytes from the network. process them as a yamux frame
	return yamux_decode(ctx->session, incoming->data, incoming->data_size);
}

int libp2p_yamux_write(void* stream_context, struct StreamMessage* message) {
	//TODO: Implement
	return 0;
}

/***
 * Check to see if there is anything waiting on the network.
 * @param stream_context the YamuxContext
 * @returns the number of bytes waiting, or -1 on error
 */
int libp2p_yamux_peek(void* stream_context) {
	if (stream_context == NULL)
		return -1;

	struct YamuxContext* ctx = (struct YamuxContext*)stream_context;
	struct Stream* parent_stream = ctx->stream->parent_stream;
	if (parent_stream == NULL)
		return -1;

	return parent_stream->peek(parent_stream->stream_context);
}

int libp2p_yamux_read_raw(void* stream_context, uint8_t* buffer, int buffer_size, int timeout_secs) {
	//TODO: Implement
	return -1;
}

struct YamuxContext* libp2p_yamux_context_new() {
	struct YamuxContext* ctx = (struct YamuxContext*) malloc(sizeof(struct YamuxContext));
	if (ctx != NULL) {
		ctx->stream = NULL;
	}
	return ctx;
}

void libp2p_yamux_stream_free(struct Stream* yamux_stream) {
	//TODO: Implement
}

int libp2p_yamux_negotiate(struct YamuxContext* ctx) {
	const char* protocolID = "/yamux/1.0.0\n";
	struct StreamMessage outgoing;
	struct StreamMessage* results = NULL;
	int retVal = 0;
	int haveTheirs = 0;
	int peek_result = 0;

	// see if they're trying to send something first
	peek_result = libp2p_yamux_peek(ctx);
	if (peek_result > 0) {
		libp2p_logger_debug("yamux", "There is %d bytes waiting for us. Perhaps it is the yamux header we're expecting.\n", peek_result);
		// get the protocol
		ctx->stream->parent_stream->read(ctx->stream->parent_stream, &results, yamux_default_timeout);
		if (results == NULL || results->data_size == 0) {
			libp2p_logger_error("yamux", "We thought we had a yamux header, but we got nothing.\n");
			goto exit;
		}
		if (strncmp((char*)results->data, protocolID, strlen(protocolID)) != 0) {
			libp2p_logger_error("yamux", "We thought we had a yamux header, but we received %d bytes that contained %s.\n", (int)results->data_size, results->data);
			goto exit;
		}
		haveTheirs = 1;
	}

	// send the protocol id
	outgoing.data = (uint8_t*)protocolID;
	outgoing.data_size = strlen(protocolID);
	if (!ctx->stream->parent_stream->write(ctx->stream->parent_stream->stream_context, &outgoing)) {
		libp2p_logger_error("yamux", "We attempted to write the yamux protocol id, but the write call failed.\n");
		goto exit;
	}

	// wait for them to send the protocol id back
	if (!haveTheirs) {
		// expect the same back
		ctx->stream->parent_stream->read(ctx->stream->parent_stream->stream_context, &results, yamux_default_timeout);
		if (results == NULL || results->data_size == 0) {
			libp2p_logger_error("yamux", "We tried to retrieve the yamux header, but we got nothing.\n");
			goto exit;
		}
		if (strncmp((char*)results->data, protocolID, strlen(protocolID)) != 0) {
			libp2p_logger_error("yamux", "We tried to retrieve the yamux header, but we received %d bytes that contained %s.\n", (int)results->data_size, results->data);
			goto exit;
		}
	}

	retVal = 1;
	exit:
	if (results != NULL)
		free(results);
	return retVal;
}

/***
 * Negotiate the Yamux protocol
 * @param parent_stream the parent stream
 * @returns a Stream initialized and ready for yamux
 */
struct Stream* libp2p_yamux_stream_new(struct Stream* parent_stream) {
	struct Stream* out = (struct Stream*)malloc(sizeof(struct Stream));
	if (out != NULL) {
		out->parent_stream = parent_stream;
		out->close = libp2p_yamux_close;
		out->read = libp2p_yamux_read;
		out->write = libp2p_yamux_write;
		out->peek = libp2p_yamux_peek;
		out->read_raw = libp2p_yamux_read_raw;
		out->address = parent_stream->address;
		// build YamuxContext
		struct YamuxContext* ctx = libp2p_yamux_context_new();
		if (ctx == NULL) {
			libp2p_yamux_stream_free(out);
			return NULL;
		}
		out->stream_context = ctx;
		ctx->stream = out;
		// attempt to negotiate yamux protocol
		if (!libp2p_yamux_negotiate(ctx)) {
			libp2p_yamux_stream_free(out);
			return NULL;
		}
	}
	return out;
}