stream.mx

这个是内存数据库中的一个管理工具

ssl_close(stream *s)
{
	struct ssl_data *ssl_data = (struct ssl_data *) s->stream_data.p;

	if (ssl_data && ssl_data->ssl) {
		int sock = SSL_shutdown(ssl_data->ssl);

		if (sock == 0)
			sock = SSL_shutdown(ssl_data->ssl);
#ifdef REPORT_SSL_SHUTDOWN_ERROR	/* usually undefined */
		if (sock == -1) {
			int err = SSL_get_error(ssl_data->ssl, sock);
			char *errstr = error(err, sock);

			fprintf(stderr, "SSL_shutdown: %s\n", errstr);
			free(errstr);
		}
#endif
		sock = SSL_get_fd(ssl_data->ssl);
		/* Related read/write (in/out, from/to) streams
		 * share a single socket which is not dup'ed (anymore)
		 * as Windows' dup doesn't work on sockets;
		 * hence, only one of the streams must/may close that
		 * socket; we choose to let the write socket do the job.
		 */
		if (s->access == ST_READ) {
			shutdown(sock, SHUT_RD);
			close(sock);
		} else {
			shutdown(sock, SHUT_WR);
		}
		if (ssl_data->ssl)
			SSL_free(ssl_data->ssl);
		ssl_data->ssl = NULL;
	}
}

static char *
ssl_error(stream *s)
{
	char *errstr;
	char buf[120];
	unsigned long e;

	assert(s->stream_data.p);
	switch (((struct ssl_data *) s->stream_data.p)->error) {
	case SSL_ERROR_ZERO_RETURN:
		errstr = "TLS/SSL connection has been closed";
		break;
	case SSL_ERROR_WANT_READ:
		errstr = "The operation did not complete (read)";
		break;
	case SSL_ERROR_WANT_WRITE:
		errstr = "The operation did not complete (write)";
		break;
	case SSL_ERROR_WANT_X509_LOOKUP:
		errstr = "The operation did not complete (X509 lookup)";
		break;
	case SSL_ERROR_WANT_CONNECT:
		errstr = "The operation did not complete (connect)";
		break;
	case SSL_ERROR_SYSCALL:
		e = ERR_get_error();
		if (e == 0) {
			if (((struct ssl_data *) s->stream_data.p)->ret == 0) {
				errstr = "EOF occurred in violation of protocol";
			} else if (((struct ssl_data *) s->stream_data.p)->ret == -1) {
				/* the underlying BIO reported an I/O error */
				errstr = "I/O error";
			} else {	/* possible? */
				errstr = "Some I/O error occurred";
			}
		} else {
			errstr = ERR_error_string(e, buf);
		}
		break;
	case SSL_ERROR_SSL:
		e = ERR_get_error();
		if (e != 0)
			errstr = ERR_error_string(e, buf);
		else {		/* possible? */
			errstr = "A failure in the SSL library occurred";
		}
		break;
	default:
		errstr = "Invalid error code";
	}
	return strdup(errstr);
}

static stream *
ssl_open(SSL * ssl, const char *name)
{
	struct ssl_data *ssl_data;
	stream *s;

	assert(ssl);
	if ((s = create_stream(name)) == NULL)
		return NULL;

	s->read = ssl_read;
	s->write = ssl_write;
	s->close = ssl_close;
	s->error = ssl_error;
	s->flush = NULL;
	s->destroy = ssl_destroy;

	if ((ssl_data = malloc(sizeof(*ssl_data))) == NULL) {
		destroy(s);
		return NULL;
	}
	ssl_data->ret = 0;
	ssl_data->error = 0;
	ssl_data->ssl = ssl;
	s->stream_data.p = (void *) ssl_data;
	return s;
}

stream *
ssl_rstream(SSL * ssl, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("ssl_rstream " PTRFMT " %s\n", PTRFMTCAST ssl, name);
#endif
	if ((s = ssl_open(ssl, name)) == NULL)
		return NULL;
	s->access = ST_READ;
	s->type = ST_BIN;
	if (s->errnr == NO__ERROR)
		ssl_read(s, (void *) &s->byteorder, sizeof(s->byteorder), 1);
	return s;
}

stream *
ssl_wstream(SSL * ssl, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("ssl_wstream " PTRFMT " %s\n", PTRFMTCAST ssl, name);
#endif
	if ((s = ssl_open(ssl, name)) == NULL)
		return NULL;
	s->access = ST_WRITE;
	s->type = ST_BIN;
	if (s->errnr == NO__ERROR)
		ssl_write(s, (void *) &s->byteorder, sizeof(s->byteorder), 1);
	return s;
}

stream *
ssl_rastream(SSL * ssl, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("ssl_rastream " PTRFMT " %s\n", PTRFMTCAST ssl, name);
#endif
	if ((s = ssl_open(ssl, name)) == NULL)
		return NULL;
	s->access = ST_READ;
	s->type = ST_ASCII;
	return s;
}

stream *
ssl_wastream(SSL * ssl, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("ssl_wastream " PTRFMT " %s\n", PTRFMTCAST ssl, name);
#endif
	if ((s = ssl_open(ssl, name)) == NULL)
		return NULL;
	s->access = ST_WRITE;
	s->type = ST_ASCII;
	return s;
}
#endif /* HAVE_OPENSSL */

static stream *
file_stream(const char *name)
{
	stream *s;

	if ((s = create_stream(name)) == NULL)
		return NULL;
	s->read = file_read;
	s->write = file_write;
	s->close = file_close;
	s->flush = file_flush;
	return s;
}

stream *
file_rstream(FILE *fp, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("file_rstream %s\n", name);
#endif
	if ((s = file_stream(name)) == NULL)
		return NULL;
	s->type = ST_BIN;
	if (fp == NULL)
		s->errnr = OPEN_ERROR;
	s->stream_data.p = (void *) fp;

	if (s->errnr == NO__ERROR) {
		fread((void *) &s->byteorder, sizeof(s->byteorder), 1, fp);
		if (ferror(fp))
			s->errnr = OPEN_ERROR;
	}
	return s;
}

stream *
file_wstream(FILE *fp, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("file_wstream %s\n", name);
#endif
	if ((s = file_stream(name)) == NULL)
		return NULL;
	s->access = ST_WRITE;
	s->type = ST_BIN;
	if (fp == NULL)
		s->errnr = OPEN_ERROR;
	s->stream_data.p = (void *) fp;

	if (s->errnr == NO__ERROR) {
		fwrite((void *) &s->byteorder, sizeof(s->byteorder), 1, fp);
		if (ferror(fp))
			s->errnr = OPEN_ERROR;
	}
	return s;
}

stream *
file_rastream(FILE *fp, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("file_rastream %s\n", name);
#endif
	if ((s = file_stream(name)) == NULL)
		return NULL;
	s->type = ST_ASCII;
	if (fp == NULL)
		s->errnr = OPEN_ERROR;
	s->stream_data.p = (void *) fp;
	return s;
}

stream *
file_wastream(FILE *fp, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("file_wastream %s\n", name);
#endif
	if ((s = file_stream(name)) == NULL)
		return NULL;
	s->access = ST_WRITE;
	s->type = ST_ASCII;
	if (fp == NULL)
		s->errnr = OPEN_ERROR;
	s->stream_data.p = (void *) fp;
	return s;
}

void
buffer_init(buffer *b, char *buf, size_t size)
{
	b->pos = 0;
	b->buf = buf;
	b->len = size;
}

buffer *
buffer_create(size_t size)
{
	buffer *b;

	if ((b = malloc(sizeof(*b))) == NULL)
		return NULL;
	b->pos = 0;
	b->buf = malloc(size);
	b->len = b->buf ? size : 0;
	return b;
}

char *
buffer_get_buf(buffer *b)
{
	char *r = b->buf;

	if (b->pos == b->len && (b->buf = realloc(b->buf, b->len+1)) == NULL) 
		return NULL;
	r[b->pos] = '\0';
	b->buf = malloc(b->len);
	b->len = b->buf ? b->len : 0;
	b->pos = 0;
	return r;
}

void
buffer_destroy(buffer *b)
{
	free(b->buf);
	free(b);
}

buffer *
stream_get_buffer(stream *s)
{
	return (buffer *) s->stream_data.p;
}

static ssize_t
buffer_read(stream *s, void *buf, size_t elmsize, size_t cnt)
{
	size_t size = elmsize * cnt;
	buffer *b = (buffer *) s->stream_data.p;

	assert(b);
	if (b->pos + size <= b->len) {
		memcpy(buf, b->buf + b->pos, size);
		b->pos += size;
		return size / elmsize;
	} else {
		s->errnr = READ_ERROR;
		return 0;
	}
}

static ssize_t
buffer_write(stream *s, const void *buf, size_t elmsize, size_t cnt)
{
	size_t size = elmsize * cnt;
	buffer *b = (buffer *) s->stream_data.p;

	assert(b);
	if (b->pos + size > b->len) {
		size_t ns = b->len;

		while (b->pos + size > ns)
			ns *= 2;
		if ((b->buf = realloc(b->buf, ns)) == NULL) {
			s->errnr = WRITE_ERROR;
			return -1;
		}
		b->len = ns;
	}
	memcpy(b->buf + b->pos, buf, size);
	b->pos += size;
	return cnt;
}

static void
buffer_close(stream *s)
{
	(void) s;
}

static int
buffer_flush(stream *s)
{
	buffer *b = (buffer *) s->stream_data.p;

	assert(b);
	b->pos = 0;
	return 0;
}

stream *
buffer_rastream(buffer *b, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("buffer_rastream %s\n", name);
#endif
	if ((s = create_stream(name)) == NULL)
		return NULL;
	s->type = ST_ASCII;
	s->read = buffer_read;
	s->write = buffer_write;
	s->close = buffer_close;
	s->flush = buffer_flush;
	s->stream_data.p = (void *) b;
	return s;
}

stream *
buffer_wastream(buffer *b, const char *name)
{
	stream *s;

#ifdef STREAM_DEBUG
	printf("buffer_wastream %s\n", name);
#endif
	if ((s = create_stream(name)) == NULL)
		return NULL;
	s->access = ST_WRITE;
	s->type = ST_ASCII;
	s->read = buffer_read;
	s->write = buffer_write;
	s->close = buffer_close;
	s->flush = buffer_flush;
	s->stream_data.p = (void *) b;
	return s;
}

/* A buffered stream consists of a sequence of blocks.  Each block
   consists of a count followed by the data in the block.  A flush is
   indicated by an empty block (i.e. just a count of 0).
 */
typedef struct bs {
	stream *s;		/* underlying stream */
	unsigned nr;		/* how far we got in buf */
	unsigned itotal;	/* amount available in current read block */
	char buf[BLOCK];	/* the buffered data (minus the size of
	                           size-short */
} bs;

static bs *
bs_create(stream *s)
{
	/* should be a binary stream */
	bs *ns;

	if ((ns = malloc(sizeof(*ns))) == NULL)
		return NULL;
	ns->s = s;
	ns->nr = 0;
	ns->itotal = 0;
	return ns;
}

/* Collect data until the internal buffer is filled, then write the
   filled buffer to the underlying stream.
   Struct field usage:
   s - the underlying stream;
   buf - the buffer in which data is collected;
   nr - how much of buf is already filled (if nr == sizeof(buf) the
        data is written to the underlying stream, so upon entry nr <
        sizeof(buf));
   itotal - unused.
 */
static ssize_t
bs_write(stream *ss, const void *buf, size_t elmsize, size_t cnt)
{
	bs *s = (bs *) ss->stream_data.p;
	size_t todo = cnt * elmsize;
	short blksize;

	assert(ss->access == ST_WRITE);
	assert(s->nr < sizeof(s->buf));
	while (todo > 0) {
		size_t n = sizeof(s->buf) - s->nr;

		if (todo < n)
			n = todo;
		memcpy(s->buf + s->nr, buf, n);
		s->nr += (unsigned) n;
		todo -= n;
		buf = (void *) ((char *) buf + n);
		if (s->nr == sizeof(s->buf)) {
			/* block is full, write it to the stream */
#ifdef BSTREAM_DEBUG
			{
				unsigned i;

				printf("W %s %u \"", ss->name, s->nr);
				for (i = 0; i < s->nr; i++)
					if (' ' <= s->buf[i] && s->buf[i] < 127)
						putchar(s->buf[i]);
					else
						printf("\\%03o", s->buf[i]);
				printf("\"\n");
			}
#endif
			/* since the block is at max BLOCK (8K) - 2 size we can
			   store it in a two byte integer */
			blksize = (short) s->nr;
			/* the last bit tells whether a flush is in there, it's not
			 * at this moment, so shift it to the left */
			blksize <<= 1;
#ifdef WORDS_BIGENDIAN
			blksize = short_int_SWAP(blksize);
#endif
			if (!stream_writeSht(s->s, blksize) || s->s->write(s->s, s->buf, 1, s->nr) != (ssize_t) s->nr) {
				ss->errnr = WRITE_ERROR;
				return -1;
			}
			s->nr = 0;
		}
	}
	return cnt;
}

/* If the internal buffer is partially filled, write it to the
   underlying stream.  Then in any case write an empty buffer to the
   underlying stream to indicate to the receiver that the data was
   flushed.
 */
static int
bs_flush(stream *ss)
{
	short blksize;
	bs *s = (bs *) ss->stream_data.p;

	assert(ss->access == ST_WRITE);
	assert(s->nr < sizeof(s->buf));
	if (ss->access == ST_WRITE) {
		/* flush the rest of buffer (if s->nr > 0), then set the
		 * last bit to 1 to to indicate user-instigated flush */
#ifdef BSTREAM_DEBUG
		if (s->nr > 0) {
			unsigned i;

			printf("W %s %u \"", ss->name, s->nr);
			for (i = 0; i < s->nr; i++)
				if (' ' <= s->buf[i] && s->buf[i] < 127)
					putchar(s->buf[i]);
				else
					printf("\\%03o", s->buf[i]);
			printf("\"\n");
			printf("W %s 0\n", ss->name);
		}
#endif
		blksize = (short) (s->nr << 1);
		/* indicate that this is the last buffer of a block by
		   setting the low-order bit */
		blksize |= (short) 1;
		/* allways flush (even empty blocks) needed for the protocol) */
#ifdef WORDS_BIGENDIAN
		blksize = short_int_SWAP(blksize);
#endif
		if ((!stream_writeSht(s->s, blksize) || s->s->write(s->s, s->buf, 1, s->nr) != (ssize_t) s->nr)) {
			ss->errnr = WRITE_ERROR;
			return -1;
		}
		s->nr = 0;
	}
	return 0;
}

/* Read buffered data and return the number of items read.  At the
   flush boundary we will return 0 to indicate the end of a block.

   Structure field usage:
   s - the underlying stream;
   buf - not used;
   itotal - the amount of data in the current block that hasn't yet
            been read;
   nr - indicates whether the flush marker has to be returned.
 */
static ssize_t
bs_read(stream *ss, void *buf, size_t elmsize, size_t cnt)
{