implementation of bitswap network receive

core/null.c

@@ -21,6 +21,7 @@
 #include "ipfs/merkledag/merkledag.h"
 #include "ipfs/merkledag/node.h"
 #include "ipfs/util/thread_pool.h"
+#include "ipfs/exchange/bitswap/network.h"
 
 #define BUF_SIZE 4096
 
@@ -101,6 +102,9 @@ int ipfs_null_marshal(const unsigned char* incoming, size_t incoming_size, struc
 		// this handles 1 transaction
 		libp2p_routing_dht_handle_message(session, connection_param->local_node->peerstore, connection_param->local_node->providerstore);
 		libp2p_logger_log("null", LOGLEVEL_DEBUG, "kademlia message handled\n");
+	} else if (protocol_compare(incoming, incoming_size, "/ipfs/bitswap/")) {
+		libp2p_logger_debug("null", "Attempting bitswap connection...\n");
+		return ipfs_bitswap_network_handle_message(connection_param->local_node, session, incoming, incoming_size);
 	}
 	else {
 		libp2p_logger_error("null", "There was a problem with this connection. It is nothing I can handle. Disconnecting.\n");

exchange/bitswap/message.c

@@ -464,7 +464,7 @@ int ipfs_bitswap_message_free(struct BitswapMessage* message) {
  * @param message the BitswapMessage
  * @returns the maximum size of the protobuf'd BitswapMessage
  */
-size_t ipfs_bitswap_message_protobuf_encode_size(struct BitswapMessage* message) {
+size_t ipfs_bitswap_message_protobuf_encode_size(const struct BitswapMessage* message) {
 	size_t total = 0;
 	if (message != NULL) {
 		if (message->blocks != NULL) {
@@ -495,7 +495,7 @@ size_t ipfs_bitswap_message_protobuf_encode_size(struct BitswapMessage* message)
  * @param bytes_written the total number of bytes written to the buffer
  * @returns true(1) on success, otherwise false(0)
  */
-int ipfs_bitswap_message_protobuf_encode(struct BitswapMessage* message, unsigned char* buffer, size_t buffer_length, size_t* bytes_written) {
+int ipfs_bitswap_message_protobuf_encode(const struct BitswapMessage* message, unsigned char* buffer, size_t buffer_length, size_t* bytes_written) {
 	size_t bytes_used = 0;
 	*bytes_written = 0;
 
@@ -558,7 +558,7 @@ int ipfs_bitswap_message_protobuf_encode(struct BitswapMessage* message, unsigne
  * @param output the newly allocated BitswapMessage
  * @returns true(1) on success, otherwise false(0)
  */
-int ipfs_bitswap_message_protobuf_decode(unsigned char* buffer, size_t buffer_length, struct BitswapMessage** output) {
+int ipfs_bitswap_message_protobuf_decode(const uint8_t* buffer, size_t buffer_length, struct BitswapMessage** output) {
 	size_t pos = 0;
 
 	*output = NULL;

exchange/bitswap/network.c

@@ -6,23 +6,7 @@
  */
 
 #include "ipfs/exchange/bitswap/network.h"
-
+#include "ipfs/exchange/bitswap/peer_request_queue.h"
-/***
- * The main loop
- */
-
-/**
- * We received a BitswapMessage from the network
- */
-/*
-ipfs_bitswap_network_receive_message(struct BitswapContext* context) {
-
-}
-*/
-
-/**
- * We want to pop something off the queue
- */
 
 /****
  * send a message to a particular peer
@@ -30,7 +14,81 @@ ipfs_bitswap_network_receive_message(struct BitswapContext* context) {
  * @param peer the peer that is the recipient
  * @param message the message to send
  */
-int ipfs_bitswap_network_send_message(const struct BitswapContext* context, const struct Libp2pPeer* peer, const struct BitswapMessage* message) {
+int ipfs_bitswap_network_send_message(const struct BitswapContext* context, struct Libp2pPeer* peer, const struct BitswapMessage* message) {
-	return 0;
+	// get a connection to the peer
+	if (peer->connection_type != CONNECTION_TYPE_CONNECTED) {
+		libp2p_peer_connect(&context->ipfsNode->identity->private_key, peer, 10);
+		if(peer->connection_type != CONNECTION_TYPE_CONNECTED)
+			return 0;
+	}
+	// protobuf the message
+	size_t buf_size = ipfs_bitswap_message_protobuf_encode_size(message);
+	uint8_t* buf = (uint8_t*) malloc(buf_size + 20);
+	if (buf == NULL)
+		return 0;
+	if (!ipfs_bitswap_message_protobuf_encode(message, &buf[20], buf_size, &buf_size)) {
+		free(buf);
+		return 0;
+	}
+	// tack on the protocol header
+	memcpy(buf, "/ipfs/bitswap/1.1.0\n", 20);
+	buf_size += 20;
+	// send it
+	int bytes_written = peer->sessionContext->default_stream->write(peer->sessionContext->default_stream, buf, buf_size);
+	if (bytes_written <= 0) {
+		free(buf);
+		return 0;
+	}
+	free(buf);
+	return 1;
 }
 
+/***
+ * Handle a raw incoming bitswap message from the network
+ * @param node us
+ * @param sessionContext the connection context
+ * @param bytes the message
+ * @param bytes_size the size of the message
+ * @returns true(1) on success, false(0) otherwise.
+ */
+int ipfs_bitswap_network_handle_message(const struct IpfsNode* node, const struct SessionContext* sessionContext, const uint8_t* bytes, size_t bytes_length) {
+	struct BitswapContext* bitswapContext = (struct BitswapContext*)node->exchange->exchangeContext;
+	// strip off the protocol header
+	int start = -1;
+	for(int i = 0; i < bytes_length; i++) {
+		if (bytes[i] == '\n') {
+			start = i+1;
+			break;
+		}
+	}
+	if (start == -1)
+		return 0;
+	// un-protobuf the message
+	struct BitswapMessage* message = NULL;
+	if (!ipfs_bitswap_message_protobuf_decode(&bytes[start], bytes_length - start, &message))
+		return 0;
+	// process the message
+	// payload - what we want
+	for(int i = 0; i < message->payload->total; i++) {
+		struct Block* blk = (struct Block*)libp2p_utils_vector_get(message->payload, i);
+		node->exchange->HasBlock(node->exchange, blk);
+	}
+	// wantlist - what they want
+	if (message->wantlist != NULL && message->wantlist->entries != NULL && message->wantlist->entries->total > 0) {
+		// get the peer
+		struct Libp2pPeer* peer = libp2p_peerstore_get_peer(node->peerstore, (unsigned char*)sessionContext->remote_peer_id, strlen(sessionContext->remote_peer_id));
+		struct PeerRequestEntry* queueEntry = ipfs_bitswap_peer_request_queue_find_entry(bitswapContext->peerRequestQueue, peer);
+		for(int i = 0; i < message->wantlist->entries->total; i++) {
+			struct WantlistEntry* entry = (struct WantlistEntry*) libp2p_utils_vector_get(message->wantlist->entries, i);
+			// turn the "block" back into a cid
+			struct Cid* cid = NULL;
+			if (!ipfs_cid_protobuf_decode(entry->block, entry->block_size, &cid)) {
+				ipfs_cid_free(cid);
+				return 0;
+			}
+			// add the cid to their queue
+			libp2p_utils_vector_add(queueEntry->current->cids, cid);
+		}
+	}
+	return 1;
+}

exchange/bitswap/peer_request_queue.c

@@ -95,7 +95,7 @@ int ipfs_bitswap_peer_request_queue_add(struct PeerRequestQueue* queue, struct P
  */
 int ipfs_bitswap_peer_request_queue_remove(struct PeerRequestQueue* queue, struct PeerRequest* request) {
 	if (request != NULL) {
-		struct PeerRequestEntry* entry = ipfs_bitswap_peer_request_queue_find_entry(queue, request);
+		struct PeerRequestEntry* entry = ipfs_bitswap_peer_request_queue_find_entry(queue, request->peer);
 		if (entry != NULL) {
 			pthread_mutex_lock(&queue->queue_mutex);
 			// remove the entry's link, and hook prior and next together
@@ -116,11 +116,11 @@ int ipfs_bitswap_peer_request_queue_remove(struct PeerRequestQueue* queue, struc
  * @param request what we're looking for
  * @returns the PeerRequestEntry or NULL if not found
  */
-struct PeerRequestEntry* ipfs_bitswap_peer_request_queue_find_entry(struct PeerRequestQueue* queue, struct PeerRequest* request) {
+struct PeerRequestEntry* ipfs_bitswap_peer_request_queue_find_entry(struct PeerRequestQueue* queue, struct Libp2pPeer* peer) {
-	if (request != NULL) {
+	if (peer != NULL) {
 		struct PeerRequestEntry* current = queue->first;
 		while (current != NULL) {
-			if (libp2p_peer_compare(current->current->peer, request->peer) == 0)
+			if (libp2p_peer_compare(current->current->peer, peer) == 0)
 				return current;
 			current = current->next;
 		}

include/ipfs/exchange/bitswap/message.h

@@ -180,7 +180,7 @@ int ipfs_bitswap_message_free(struct BitswapMessage* message);
  * @param message the BitswapMessage
  * @returns the maximum size of the protobuf'd BitswapMessage
  */
-size_t ipfs_bitswap_message_protobuf_encode_size(struct BitswapMessage* message);
+size_t ipfs_bitswap_message_protobuf_encode_size(const struct BitswapMessage* message);
 
 /***
  * Encode a BitswapMessage into a protobuf buffer
@@ -190,7 +190,7 @@ size_t ipfs_bitswap_message_protobuf_encode_size(struct BitswapMessage* message)
  * @param bytes_written the total number of bytes written to the buffer
  * @returns true(1) on success, otherwise false(0)
  */
-int ipfs_bitswap_message_protobuf_encode(struct BitswapMessage* message, unsigned char* buffer, size_t buffer_length, size_t* bytes_written);
+int ipfs_bitswap_message_protobuf_encode(const struct BitswapMessage* message, unsigned char* buffer, size_t buffer_length, size_t* bytes_written);
 
 /***
  * Decode a BitswapMessage from a protobuf
@@ -199,7 +199,7 @@ int ipfs_bitswap_message_protobuf_encode(struct BitswapMessage* message, unsigne
  * @param output the newly allocated BitswapMessage
  * @returns true(1) on success, otherwise false(0)
  */
-int ipfs_bitswap_message_protobuf_decode(unsigned char* buffer, size_t buffer_length, struct BitswapMessage** output);
+int ipfs_bitswap_message_protobuf_decode(const uint8_t* buffer, size_t buffer_length, struct BitswapMessage** output);
 
 /****
  * Add a vector of Cids to the bitswap message

include/ipfs/exchange/bitswap/network.h

@@ -70,4 +70,14 @@ struct BitswapNetwork {
  * @param peer the peer that is the recipient
  * @param message the message to send
  */
-int ipfs_bitswap_network_send_message(const struct BitswapContext* context, const struct Libp2pPeer* peer, const struct BitswapMessage* message);
+int ipfs_bitswap_network_send_message(const struct BitswapContext* context, struct Libp2pPeer* peer, const struct BitswapMessage* message);
+
+/***
+ * Handle a raw incoming bitswap message from the network
+ * @param node us
+ * @param sessionContext the connection context
+ * @param bytes the message
+ * @param bytes_size the size of the message
+ * @returns true(1) on success, false(0) otherwise.
+ */
+int ipfs_bitswap_network_handle_message(const struct IpfsNode* node, const struct SessionContext* sessionContext, const uint8_t* bytes, size_t bytes_length);

include/ipfs/exchange/bitswap/peer_request_queue.h

@@ -83,7 +83,7 @@ struct PeerRequest* ipfs_bitswap_peer_request_queue_pop(struct PeerRequestQueue*
  * @param request what we're looking for
  * @returns the PeerRequestEntry or NULL if not found
  */
-struct PeerRequestEntry* ipfs_bitswap_peer_request_queue_find_entry(struct PeerRequestQueue* queue, struct PeerRequest* request);
+struct PeerRequestEntry* ipfs_bitswap_peer_request_queue_find_entry(struct PeerRequestQueue* queue, struct Libp2pPeer* peer);
 
 /***
  * Add a block to the appropriate peer's queue