1

#include "msg_proto.h"

     2

     3

#include <string.h>

     4

#include <stdint.h>

     5

#include <arpa/inet.h>

     6

     7

/**

     8

 * I/O buffer

     9

 */

    10

struct msg_buf {

    11

    /** Buffer base pointer */

    12

    char *base;

    13

    14

    /** Size of the buffer */

    15

    size_t size;

    16

    17

    /** Current read/write offset */

    18

    size_t off;

    19

};

    20

    21

/**

    22

 * The minimum size used for any msg_buf::size related operation.

    23

 */

    24

#define MSG_BUF_MIN_SIZE 1024

    25

    26

/**

    27

 * Growth rate for size

    28

 */

    29

#define MSG_BUF_GROW_RATE 2

    30

    31

/**

    32

 * Initialize a message buffer at the given initial size

    33

 */

    34

err_t msg_buf_init (struct msg_buf *buf, size_t hint)

    35

{

    36

    // apply minimum size

    37

    if (hint < MSG_BUF_MIN_SIZE)

    38

        hint = MSG_BUF_MIN_SIZE;

    39

    40

    // allocate the initial buffer

    41

    if ((buf->base = malloc(hint)) == NULL)

    42

        return ERR_MEM;

    43

    44

    // set fields

    45

    buf->size = hint;

    46

    buf->off = 0;

    47

    48

    // ok

    49

    return SUCCESS;

    50

}

    51

    52

/**

    53

 * Grow the buffer if needed to fit the given capacity.

    54

 */

    55

err_t msg_buf_grow (struct msg_buf *buf, size_t size)

    56

{

    57

    char *tmp = buf->base;

    58

    59

    if (buf->size >= size)

    60

        // nothing to do

    61

        return SUCCESS;

    62

    63

    // calculate new size

    64

    while (buf->size < size)

    65

        buf->size *= MSG_BUF_GROW_RATE;

    66

    67

    // resize

    68

    if ((buf->base = realloc(buf->base, buf->size)) == NULL) {

    69

        buf->base = tmp;

    70

    71

        return ERR_MEM;

    72

    }

    73

    74

    // ok

    75

    return SUCCESS;

    76

}

    77

    78

/**

    79

 * Drain \a len bytes off the head of the buffer

    80

 */

    81

err_t msg_buf_drain (struct msg_buf *buf, size_t len)

    82

{

    83

    // simple memmove

    84

    memmove(buf->base, buf->base + len, buf->off - len);

    85

    86

    // update offfset

    87

    buf->off -= len;

    88

    89

    // ok

    90

    return SUCCESS;

    91

}

    92

    93

/**

    94

 * Read into the buffer from a transport_t.

    95

 *

    96

 * This will attempt to read \a len bytes onto the end of the buffer, growing it if needed to fit.

    97

 *

    98

 * This may read/return more data than the given len. Use msg_buf_drain the remove the data from the buffer once you

    99

 * have used it.

   100

 *

   101

 * Returns the number of new bytes read, zero for transport read buffer empty, -err_t for error.

   102

 */

   103

ssize_t msg_buf_read (struct msg_buf *buf, transport_t *transport, size_t len, error_t *err)

   104

{

   105

    ssize_t ret;

   106

   107

    // clamp size

   108

    if (len < MSG_BUF_MIN_SIZE)

   109

        len = MSG_BUF_MIN_SIZE;

   110

   111

    // ensure space

   112

    if ((ERROR_CODE(err) = msg_buf_grow(buf, buf->off + len)))

   113

        goto error;

   114

   115

    // read

   116

    if ((ret = transport_read(transport, buf->base + buf->off, len, err)) < 0)

   117

        goto error;

   118

   119

    // no data left?

   120

    if (!ret)

   121

        return 0;

   122

   123

    // update offset

   124

    buf->off += ret;

   125

   126

    // ok

   127

    return ret;

   128

   129

error:

   130

    return -ERROR_CODE(err);    

   131

}

   132

   133

/**

   134

 * Drives transport_write on the given data until all the given data is written, or zero is returned.

   135

 *

   136

 * @param transport transport to write to

   137

 * @param data input data

   138

 * @param len number of bytes to write from data

   139

 * @param err returned error info

   140

 * @return number of bytes written (which may be zero or less than len), or -err_t.

   141

 */

   142

static ssize_t _transport_write_all (transport_t *transport, const char *data, size_t len, error_t *err)

   143

{

   144

    ssize_t ret;

   145

    size_t written = 0;

   146

   147

    while (len) {

   148

        // try and write out remaining data

   149

        if ((ret = transport_write(transport, data, len, err)) < 0)

   150

            goto error;

   151

   152

        if (!ret) {

   153

            // write buffer full

   154

            break;

   155

   156

        } else { 

   157

            // update and continue

   158

            written += ret;

   159

            data += ret;

   160

            len -= ret;

   161

        }

   162

    }

   163

   164

    // ok

   165

    return written;

   166

   167

error:

   168

    return -ERROR_CODE(err);    

   169

}

   170

   171

/**

   172

 * If the buffer is empty, this will attempt to write the given data directly using transport_write until either all

   173

 * the data is written (in which case nothing more needs to be done), or the transport won't accept any more writes,

   174

 * in which case the remaining data will be buffered.

   175

 *

   176

 * If the buffer is not empty, then the given data will be added to the end of the buffer, since otherwise the order of

   177

 * data would be broken.

   178

 *

   179

 * In either case, transport_write semantics garuntee that our buffer will either be empty, or an on_write will be

   180

 * pending on the transport. See msg_buf_flush() for how to handle transport_callbacks::on_write.

   181

 */

   182

err_t msg_buf_write (struct msg_buf *buf, transport_t *transport, const void *data_ptr, size_t len, error_t *err)

   183

{

   184

    ssize_t ret;

   185

    const char *data = data_ptr;

   186

   187

    if (!buf->off) {

   188

        // no data buffered, so we can try and write directly

   189

        if ((ret = _transport_write_all(transport, data, len, err)) < 0)

   190

            goto error;

   191

   192

        // update written

   193

        data += ret;

   194

        len -= ret;

   195

   196

        if (len == 0)

   197

            // wrote it all

   198

            return SUCCESS;

   199

    }

   200

   201

    // ensure space

   202

    if ((ERROR_CODE(err) = msg_buf_grow(buf, buf->off + len)))

   203

        goto error;

   204

   205

    // store

   206

    memcpy(buf->base + buf->off, data, len);

   207

   208

    // update

   209

    buf->off += len;

   210

   211

    // ok

   212

    return SUCCESS;

   213

   214

error:

   215

    return ERROR_CODE(err);    

   216

}

   217

   218

/**

   219

 * Flush buffered write data to the transport, driving transport_write() until either all of our bufferd data has been

   220

 * written, or the transport will not accept any more.

   221

 *

   222

 * In either case, transport_write semantics garuntee that our buffer will either be empty, or an on_write will be

   223

 * pending on the transport.

   224

 */

   225

err_t msg_buf_flush (struct msg_buf *buf, transport_t *transport, error_t *err)

   226

{

   227

    ssize_t ret;

   228

   229

    // write

   230

    if ((ret = _transport_write_all(transport, buf->base, buf->off, err)) < 0)

   231

        goto error;

   232

   233

    if (ret)

   234

        // unbuffer the written data

   235

        msg_buf_drain(buf, ret);

   236

   237

    // ok

   238

    return SUCCESS;

   239

   240

error:

   241

    return ERROR_CODE(err);    

   242

}

   243

   244

/**

   245

 * Deinitialize msg_buf to release allocated buffers

   246

 */

   247

void msg_buf_deinit (struct msg_buf *buf)

   248

{

   249

    // release

   250

    free(buf->base);

   251

   252

    // reset

   253

    buf->base = NULL;

   254

    buf->size = buf->off = 0;

   255

}

   256

   257

/**

   258

 * Message header

   259

 */

   260

struct msg_header {

   261

    /** Message length, including header */

   262

    uint16_t len;

   263

};

   264

   265

/**

   266

 * Message header size

   267

 */

   268

#define MSG_PROTO_HEADER_SIZE (sizeof(uint16_t))

   269

   270

/**

   271

 * Our state struct

   272

 */

   273

struct msg_proto {

   274

    /** The transport */

   275

    transport_t *transport;

   276

   277

    /** User callbacks */

   278

    const struct msg_proto_callbacks *cb_tbl;

   279

   280

    /** User callback argument */

   281

    void *cb_arg;

   282

   283

    /** Input buffer */

   284

    struct msg_buf in;

   285

   286

    /** Output buffer */

   287

    struct msg_buf out;

   288

};

   289

   290

/**

   291

 * Signal error to user

   292

 */

   293

static void msg_proto_error (struct msg_proto *proto, const error_t *err)

   294

{

   295

    // invoke user callback

   296

    proto->cb_tbl->on_error(proto, err, proto->cb_arg);

   297

}

   298

   299

/**

   300

 * Attempt to read the current header from our input buffer.

   301

 *

   302

 * Returns >0 for full header, 0 for incomplete header, -err_t for error.

   303

 */

   304

static int msg_proto_peek_header (struct msg_proto *proto, struct msg_header *header, error_t *err)

   305

{

   306

    if (proto->in.off < MSG_PROTO_HEADER_SIZE)

   307

        // not enough data for header

   308

        return 0;

   309

   310

    // read header

   311

    header->len = ntohs(*((uint16_t *) proto->in.base));

   312

   313

    // bad header?

   314

    if (header->len < MSG_PROTO_HEADER_SIZE)

   315

        JUMP_SET_ERROR_STR(err, ERR_MISC, "message_header::len");

   316

   317

    // ok, got header

   318

    return 1;

   319

   320

error:

   321

    return -ERROR_CODE(err);    

   322

}

   323

   324

/**

   325

 * Recieved a message with the given header, and a pointer to the message data

   326

 *

   327

 * XXX: what to do if the user callback destroys the msg_proto?

   328

 */

   329

static err_t msg_proto_on_msg (struct msg_proto *proto, struct msg_header *header, char *data, error_t *err)

   330

{

   331

    (void) err;

   332

   333

    // invoke user callback

   334

    proto->cb_tbl->on_msg(proto, data, header->len - MSG_PROTO_HEADER_SIZE, proto->cb_arg);

   335

   336

    // XXX: handle user errors

   337

    return SUCCESS;

   338

}

   339

   340

static void msg_proto_on_read (transport_t *transport, void *arg)

   341

{

   342

    struct msg_proto *proto = arg;

   343

    struct msg_header header;

   344

    ssize_t ret;

   345

    error_t err;

   346

   347

    // we might be able to read more than one message per event

   348

    do {

   349

        // try and read message length for incomplete message

   350

        if ((ret = msg_proto_peek_header(proto, &header, &err)) < 0)

   351

            goto error;

   352

   353

        // need to read more data?

   354

        if (!ret || header.len > proto->in.off) {

   355

            // msg_buf_read a minimum size, so passing a zero is OK

   356

            size_t to_read = ret ? header.len : 0;

   357

   358

            // read into our buffer

   359

            if ((ret = msg_buf_read(&proto->in, transport, to_read, &err)) < 0)

   360

                goto error;

   361

   362

        } else {

   363

            // handle full message

   364

            if (msg_proto_on_msg(proto, &header, proto->in.base + MSG_PROTO_HEADER_SIZE, &err))

   365

                goto error;

   366

   367

            // remove the data from the buffer

   368

            msg_buf_drain(&proto->in, header.len);

   369

        }

   370

    } while (ret);

   371

   372

    // ok

   373

    return;

   374

   375

error:

   376

    // notify user

   377

    msg_proto_error(proto, &err);    

   378

}

   379

   380

static void msg_proto_on_write (transport_t *transport, void *arg)

   381

{

   382

    struct msg_proto *proto = arg;

   383

    error_t err;

   384

   385

    // flush

   386

    if (msg_buf_flush(&proto->out, transport, &err))

   387

        // notify user on transport errors

   388

        msg_proto_error(proto, &err);

   389

}

   390

   391

static void msg_proto_on_error (transport_t *transport, const error_t *err, void *arg)

   392

{

   393

    struct msg_proto *proto = arg;

   394

   395

    (void) transport;

   396

   397

    // report to user

   398

    msg_proto_error(proto, err);

   399

}

   400

   401

static const struct transport_callbacks msg_proto_transport_callbacks = {

   402

    .on_read    = msg_proto_on_read,

   403

    .on_write   = msg_proto_on_write,

   404

    .on_error   = msg_proto_on_error,

   405

};

   406

   407

err_t msg_proto_create (struct msg_proto **proto_ptr, transport_t *transport, const struct msg_proto_callbacks *cb_tbl, void *cb_arg, error_t *err)

   408

{

   409

    struct msg_proto *proto;

   410

   411

    // alloc

   412

    if ((proto = calloc(1, sizeof(*proto))) == NULL)

   413

        return ERR_MEM;

   414

   415

    // store

   416

    proto->transport = transport;

   417

    proto->cb_tbl = cb_tbl;

   418

    proto->cb_arg = cb_arg;

   419

   420

    // init

   421

    if (

   422

            (ERROR_CODE(err) = msg_buf_init(&proto->in, 0))

   423

        ||  (ERROR_CODE(err) = msg_buf_init(&proto->out, 0))

   424

    )

   425

        goto error;

   426

   427

    // setup transport

   428

    if ((ERROR_CODE(err) = transport_events(transport, TRANSPORT_READ | TRANSPORT_WRITE)))

   429

        goto error;

   430

   431

    transport_set_callbacks(transport, &msg_proto_transport_callbacks, proto);

   432

   433

    // ok

   434

    *proto_ptr = proto;

   435

   436

    return SUCCESS;

   437

   438

error:

   439

    // release

   440

    msg_proto_destroy(proto);

   441

   442

    return ERROR_CODE(err);

   443

}

   444

   445

/**

   446

 * Build and write out the data for the given header

   447

 */

   448

static err_t msg_proto_write_header (struct msg_proto *proto, const struct msg_header *header, error_t *err)

   449

{

   450

    char buf[MSG_PROTO_HEADER_SIZE];

   451

   452

    // validate

   453

    if (header->len < MSG_PROTO_HEADER_SIZE)

   454

        return SET_ERROR(err, ERR_MISC);

   455

   456

    // build

   457

    *((uint16_t *) buf) = htons(header->len);

   458

   459

    // write

   460

    return msg_buf_write(&proto->out, proto->transport, buf, sizeof(buf), err);

   461

}

   462

   463

err_t msg_proto_send (struct msg_proto *proto, const void *data, size_t len, error_t *err)

   464

{

   465

    struct msg_header header;

   466

   467

    // build header

   468

    header.len = MSG_PROTO_HEADER_SIZE + len;

   469

   470

    // write it

   471

    if (

   472

            msg_proto_write_header(proto, &header, err)

   473

        ||  msg_buf_write(&proto->out, proto->transport, data, len, err)

   474

    )

   475

        return ERROR_CODE(err);

   476

   477

    // ok

   478

    return SUCCESS;

   479

}

   480

   481

void msg_proto_destroy (struct msg_proto *proto)

   482

{

   483

    // drop buffers

   484

    msg_buf_deinit(&proto->in);

   485

    msg_buf_deinit(&proto->out);

   486

   487

    // kill transport

   488

    transport_destroy(proto->transport);

   489

   490

    // release ourself

   491

    free(proto);

   492

}

   493