SocketBuffer.c

Go to the documentation of this file.

/*******************************************************************************
 * Copyright (c) 2009, 2020 IBM Corp.
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v2.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *    https://www.eclipse.org/legal/epl-2.0/
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Ian Craggs - initial API and implementation and/or initial documentation
 *    Ian Craggs, Allan Stockdill-Mander - SSL updates
 *    Ian Craggs - fix for issue #244, issue #20
 *******************************************************************************/

#include "SocketBuffer.h"
#include "LinkedList.h"
#include "Log.h"
#include "Messages.h"
#include "StackTrace.h"

#include <stdlib.h>
#include <stdio.h>
#include <string.h>

#include "Heap.h"

#if defined(_WIN32) || defined(_WIN64)
#define iov_len len
#define iov_base buf
#endif

static socket_queue* def_queue;

static List* queues;

static List writes;


int socketcompare(void* a, void* b);
int SocketBuffer_newDefQ(void);
void SocketBuffer_freeDefQ(void);
int pending_socketcompare(void* a, void* b);


int socketcompare(void* a, void* b)
{
        return ((socket_queue*)a)->socket == *(int*)b;
}


int SocketBuffer_newDefQ(void)
{
        int rc = PAHO_MEMORY_ERROR;

        def_queue = malloc(sizeof(socket_queue));
        if (def_queue)
        {
                def_queue->buflen = 1000;
                def_queue->buf = malloc(def_queue->buflen);
                if (def_queue->buf)
                {
                        def_queue->socket = def_queue->index = 0;
                        def_queue->buflen = def_queue->datalen = def_queue->headerlen = 0;
                        rc = 0;
                }
        }
        return rc;
}


int SocketBuffer_initialize(void)
{
        int rc = 0;

        FUNC_ENTRY;
        rc = SocketBuffer_newDefQ();
        if (rc == 0)
        {
                if ((queues = ListInitialize()) == NULL)
                        rc = PAHO_MEMORY_ERROR;
        }
        ListZero(&writes);
        FUNC_EXIT_RC(rc);
        return rc;
}


void SocketBuffer_freeDefQ(void)
{
        free(def_queue->buf);
        free(def_queue);
        def_queue = NULL;
}


void SocketBuffer_terminate(void)
{
        ListElement* cur = NULL;
        ListEmpty(&writes);

        FUNC_ENTRY;
        while (ListNextElement(queues, &cur))
                free(((socket_queue*)(cur->content))->buf);
        ListFree(queues);
        SocketBuffer_freeDefQ();
        FUNC_EXIT;
}


void SocketBuffer_cleanup(int socket)
{
        FUNC_ENTRY;
        SocketBuffer_writeComplete(socket); /* clean up write buffers */
        if (ListFindItem(queues, &socket, socketcompare))
        {
                free(((socket_queue*)(queues->current->content))->buf);
                ListRemove(queues, queues->current->content);
        }
        if (def_queue->socket == socket)
        {
                def_queue->socket = def_queue->index = 0;
                def_queue->headerlen = def_queue->datalen = 0;
        }
        FUNC_EXIT;
}


char* SocketBuffer_getQueuedData(int socket, size_t bytes, size_t* actual_len)
{
        socket_queue* queue = NULL;

        FUNC_ENTRY;
        if (ListFindItem(queues, &socket, socketcompare))
        {  /* if there is queued data for this socket, add any data read to it */
                queue = (socket_queue*)(queues->current->content);
                *actual_len = queue->datalen;
        }
        else
        {
                *actual_len = 0;
                queue = def_queue;
        }
        if (bytes > queue->buflen)
        {
                if (queue->datalen > 0)
                {
                        void* newmem = malloc(bytes);

                        free(queue->buf);
                        queue->buf = newmem;
                        if (!newmem)
                                goto exit;
                        memcpy(newmem, queue->buf, queue->datalen);
                }
                else
                        queue->buf = realloc(queue->buf, bytes);
                queue->buflen = bytes;
        }
exit:
        FUNC_EXIT;
        return queue->buf;
}


int SocketBuffer_getQueuedChar(int socket, char* c)
{
        int rc = SOCKETBUFFER_INTERRUPTED;

        FUNC_ENTRY;
        if (ListFindItem(queues, &socket, socketcompare))
        {  /* if there is queued data for this socket, read that first */
                socket_queue* queue = (socket_queue*)(queues->current->content);
                if (queue->index < queue->headerlen)
                {
                        *c = queue->fixed_header[(queue->index)++];
                        Log(TRACE_MAX, -1, "index is now %d, headerlen %d", queue->index, (int)queue->headerlen);
                        rc = SOCKETBUFFER_COMPLETE;
                        goto exit;
                }
                else if (queue->index > 4)
                {
                        Log(LOG_FATAL, -1, "header is already at full length");
                        rc = SOCKET_ERROR;
                        goto exit;
                }
        }
exit:
        FUNC_EXIT_RC(rc);
        return rc;  /* there was no queued char if rc is SOCKETBUFFER_INTERRUPTED*/
}


void SocketBuffer_interrupted(int socket, size_t actual_len)
{
        socket_queue* queue = NULL;

        FUNC_ENTRY;
        if (ListFindItem(queues, &socket, socketcompare))
                queue = (socket_queue*)(queues->current->content);
        else /* new saved queue */
        {
                queue = def_queue;
                /* if SocketBuffer_queueChar() has not yet been called, then the socket number
                  in def_queue will not have been set.  Issue #244.
                  If actual_len == 0 then we may not need to do anything - I'll leave that
                  optimization for another time. */
                queue->socket = socket;
                ListAppend(queues, def_queue, sizeof(socket_queue)+def_queue->buflen);
                SocketBuffer_newDefQ();
        }
        queue->index = 0;
        queue->datalen = actual_len;
        FUNC_EXIT;
}


char* SocketBuffer_complete(int socket)
{
        FUNC_ENTRY;
        if (ListFindItem(queues, &socket, socketcompare))
        {
                socket_queue* queue = (socket_queue*)(queues->current->content);
                SocketBuffer_freeDefQ();
                def_queue = queue;
                ListDetach(queues, queue);
        }
        def_queue->socket = def_queue->index = 0;
        def_queue->headerlen = def_queue->datalen = 0;
        FUNC_EXIT;
        return def_queue->buf;
}


void SocketBuffer_queueChar(int socket, char c)
{
        int error = 0;
        socket_queue* curq = def_queue;

        FUNC_ENTRY;
        if (ListFindItem(queues, &socket, socketcompare))
                curq = (socket_queue*)(queues->current->content);
        else if (def_queue->socket == 0)
        {
                def_queue->socket = socket;
                def_queue->index = 0;
                def_queue->datalen = 0;
        }
        else if (def_queue->socket != socket)
        {
                Log(LOG_FATAL, -1, "attempt to reuse socket queue");
                error = 1;
        }
        if (curq->index > 4)
        {
                Log(LOG_FATAL, -1, "socket queue fixed_header field full");
                error = 1;
        }
        if (!error)
        {
                curq->fixed_header[(curq->index)++] = c;
                curq->headerlen = curq->index;
        }
        Log(TRACE_MAX, -1, "queueChar: index is now %d, headerlen %d", curq->index, (int)curq->headerlen);
        FUNC_EXIT;
}


#if defined(OPENSSL)
int SocketBuffer_pendingWrite(int socket, SSL* ssl, int count, iobuf* iovecs, int* frees, size_t total, size_t bytes)
#else
int SocketBuffer_pendingWrite(int socket, int count, iobuf* iovecs, int* frees, size_t total, size_t bytes)
#endif
{
        int i = 0;
        pending_writes* pw = NULL;
        int rc = 0;

        FUNC_ENTRY;
        /* store the buffers until the whole packet is written */
        if ((pw = malloc(sizeof(pending_writes))) == NULL)
        {
                rc = PAHO_MEMORY_ERROR;
                goto exit;
        }
        pw->socket = socket;
#if defined(OPENSSL)
        pw->ssl = ssl;
#endif
        pw->bytes = bytes;
        pw->total = total;
        pw->count = count;
        for (i = 0; i < count; i++)
        {
                pw->iovecs[i] = iovecs[i];
                pw->frees[i] = frees[i];
        }
        ListAppend(&writes, pw, sizeof(pw) + total);
exit:
        FUNC_EXIT_RC(rc);
        return rc;
}


int pending_socketcompare(void* a, void* b)
{
        return ((pending_writes*)a)->socket == *(int*)b;
}


pending_writes* SocketBuffer_getWrite(int socket)
{
        ListElement* le = ListFindItem(&writes, &socket, pending_socketcompare);
        return (le) ? (pending_writes*)(le->content) : NULL;
}


int SocketBuffer_writeComplete(int socket)
{
        return ListRemoveItem(&writes, &socket, pending_socketcompare);
}


pending_writes* SocketBuffer_updateWrite(int socket, char* topic, char* payload)
{
        pending_writes* pw = NULL;
        ListElement* le = NULL;

        FUNC_ENTRY;
        if ((le = ListFindItem(&writes, &socket, pending_socketcompare)) != NULL)
        {
                pw = (pending_writes*)(le->content);
                if (pw->count == 4)
                {
                        pw->iovecs[2].iov_base = topic;
                        pw->iovecs[3].iov_base = payload;
                }
        }

        FUNC_EXIT;
        return pw;
}