summaryrefslogtreecommitdiffstats
path: root/runtime/stream.c
diff options
context:
space:
mode:
Diffstat (limited to 'runtime/stream.c')
-rw-r--r--runtime/stream.c934
1 files changed, 934 insertions, 0 deletions
diff --git a/runtime/stream.c b/runtime/stream.c
new file mode 100644
index 00000000..1be4571a
--- /dev/null
+++ b/runtime/stream.c
@@ -0,0 +1,934 @@
+//TODO: O_TRUC mode!
+/* The serial stream class.
+ *
+ * A serial stream provides serial data access. In theory, serial streams
+ * can be implemented via a number of methods (e.g. files or in-memory
+ * streams). In practice, there currently only exist the file type (aka
+ * "driver").
+ *
+ * File begun on 2008-01-09 by RGerhards
+ *
+ * Copyright 2008 Rainer Gerhards and Adiscon GmbH.
+ *
+ * This file is part of the rsyslog runtime library.
+ *
+ * The rsyslog runtime library is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Lesser General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * The rsyslog runtime library is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public License
+ * along with the rsyslog runtime library. If not, see <http://www.gnu.org/licenses/>.
+ *
+ * A copy of the GPL can be found in the file "COPYING" in this distribution.
+ * A copy of the LGPL can be found in the file "COPYING.LESSER" in this distribution.
+ */
+#include "config.h"
+
+#include <stdlib.h>
+#include <string.h>
+#include <assert.h>
+#include <signal.h>
+#include <pthread.h>
+#include <fcntl.h>
+#include <unistd.h>
+#include <sys/stat.h> /* required for HP UX */
+#include <errno.h>
+
+#include "rsyslog.h"
+#include "syslogd.h"
+#include "stringbuf.h"
+#include "srUtils.h"
+#include "obj.h"
+#include "stream.h"
+
+/* static data */
+DEFobjStaticHelpers
+
+/* methods */
+
+/* first, we define type-specific handlers. The provide a generic functionality,
+ * but for this specific type of strm. The mapping to these handlers happens during
+ * strm construction. Later on, handlers are called by pointers present in the
+ * strm instance object.
+ */
+
+/* open a strm file
+ * It is OK to call this function when the stream is already open. In that
+ * case, it returns immediately with RS_RET_OK
+ */
+static rsRetVal strmOpenFile(strm_t *pThis)
+{
+ DEFiRet;
+ int iFlags;
+
+ ASSERT(pThis != NULL);
+ ASSERT(pThis->tOperationsMode == STREAMMODE_READ || pThis->tOperationsMode == STREAMMODE_WRITE);
+
+ if(pThis->fd != -1)
+ ABORT_FINALIZE(RS_RET_OK);
+
+ if(pThis->pszFName == NULL)
+ ABORT_FINALIZE(RS_RET_FILE_PREFIX_MISSING);
+
+ if(pThis->sType == STREAMTYPE_FILE_CIRCULAR) {
+ CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir,
+ pThis->pszFName, pThis->lenFName, pThis->iCurrFNum, pThis->iFileNumDigits));
+ } else {
+ if(pThis->pszDir == NULL) {
+ if((pThis->pszCurrFName = (uchar*) strdup((char*) pThis->pszFName)) == NULL)
+ ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+ } else {
+ CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir,
+ pThis->pszFName, pThis->lenFName, -1, 0));
+ }
+ }
+
+ /* compute which flags we need to provide to open */
+ if(pThis->tOperationsMode == STREAMMODE_READ)
+ iFlags = O_RDONLY;
+ else
+ iFlags = O_WRONLY | O_CREAT;
+
+ iFlags |= pThis->iAddtlOpenFlags;
+
+ pThis->fd = open((char*)pThis->pszCurrFName, iFlags, pThis->tOpenMode);
+ if(pThis->fd == -1) {
+ int ierrnoSave = errno;
+ dbgoprint((obj_t*) pThis, "open error %d, file '%s'\n", errno, pThis->pszCurrFName);
+ if(ierrnoSave == ENOENT)
+ ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND);
+ else
+ ABORT_FINALIZE(RS_RET_IO_ERROR);
+ }
+
+ pThis->iCurrOffs = 0;
+
+ dbgoprint((obj_t*) pThis, "opened file '%s' for %s (0x%x) as %d\n", pThis->pszCurrFName,
+ (pThis->tOperationsMode == STREAMMODE_READ) ? "READ" : "WRITE", iFlags, pThis->fd);
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* close a strm file
+ * Note that the bDeleteOnClose flag is honored. If it is set, the file will be
+ * deleted after close. This is in support for the qRead thread.
+ */
+static rsRetVal strmCloseFile(strm_t *pThis)
+{
+ DEFiRet;
+
+ ASSERT(pThis != NULL);
+ ASSERT(pThis->fd != -1);
+ dbgoprint((obj_t*) pThis, "file %d closing\n", pThis->fd);
+
+ if(pThis->tOperationsMode == STREAMMODE_WRITE)
+ strmFlush(pThis);
+
+ close(pThis->fd); // TODO: error check
+ pThis->fd = -1;
+
+ if(pThis->bDeleteOnClose) {
+ unlink((char*) pThis->pszCurrFName); // TODO: check returncode
+ }
+
+ pThis->iCurrOffs = 0; /* we are back at begin of file */
+ if(pThis->pszCurrFName != NULL) {
+ free(pThis->pszCurrFName); /* no longer needed in any case (just for open) */
+ pThis->pszCurrFName = NULL;
+ }
+
+ RETiRet;
+}
+
+
+/* switch to next strm file
+ * This method must only be called if we are in a multi-file mode!
+ */
+static rsRetVal
+strmNextFile(strm_t *pThis)
+{
+ DEFiRet;
+
+ ASSERT(pThis != NULL);
+ ASSERT(pThis->iMaxFiles != 0);
+ ASSERT(pThis->fd != -1);
+
+ CHKiRet(strmCloseFile(pThis));
+
+ /* we do modulo operation to ensure we obey the iMaxFile property. This will always
+ * result in a file number lower than iMaxFile, so it if wraps, the name is back to
+ * 0, which results in the first file being overwritten. Not desired for queues, so
+ * make sure their iMaxFiles is large enough. But it is well-desired for other
+ * use cases, e.g. a circular output log file. -- rgerhards, 2008-01-10
+ */
+ pThis->iCurrFNum = (pThis->iCurrFNum + 1) % pThis->iMaxFiles;
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* handle the eof case for monitored files.
+ * If we are monitoring a file, someone may have rotated it. In this case, we
+ * also need to close it and reopen it under the same name.
+ * rgerhards, 2008-02-13
+ */
+static rsRetVal
+strmHandleEOFMonitor(strm_t *pThis)
+{
+ DEFiRet;
+ struct stat statOpen;
+ struct stat statName;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+ /* find inodes of both current descriptor as well as file now in file
+ * system. If they are different, the file has been rotated (or
+ * otherwise rewritten). We also check the size, because the inode
+ * does not change if the file is truncated (this, BTW, is also a case
+ * where we actually loose log lines, because we can not do anything
+ * against truncation...). We do NOT rely on the time of last
+ * modificaton because that may not be available under all
+ * circumstances. -- rgerhards, 2008-02-13
+ */
+ if(fstat(pThis->fd, &statOpen) == -1)
+ ABORT_FINALIZE(RS_RET_IO_ERROR);
+ if(stat((char*) pThis->pszCurrFName, &statName) == -1)
+ ABORT_FINALIZE(RS_RET_IO_ERROR);
+ if(statOpen.st_ino == statName.st_ino && pThis->iCurrOffs == statName.st_size) {
+ ABORT_FINALIZE(RS_RET_EOF);
+ } else {
+ /* we had a file change! */
+ CHKiRet(strmCloseFile(pThis));
+ CHKiRet(strmOpenFile(pThis));
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* handle the EOF case of a stream
+ * The EOF case is somewhat complicated, as the proper action depends on the
+ * mode the stream is in. If there are multiple files (circular logs, most
+ * important use case is queue files!), we need to close the current file and
+ * try to open the next one.
+ * rgerhards, 2008-02-13
+ */
+static rsRetVal
+strmHandleEOF(strm_t *pThis)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+ switch(pThis->sType) {
+ case STREAMTYPE_FILE_SINGLE:
+ ABORT_FINALIZE(RS_RET_EOF);
+ break;
+ case STREAMTYPE_FILE_CIRCULAR:
+ /* we have multiple files and need to switch to the next one */
+ /* TODO: think about emulating EOF in this case (not yet needed) */
+#if 0
+ if(pThis->iMaxFiles == 0) /* TODO: why do we need this? ;) */
+ ABORT_FINALIZE(RS_RET_EOF);
+#endif
+ dbgoprint((obj_t*) pThis, "file %d EOF\n", pThis->fd);
+ CHKiRet(strmNextFile(pThis));
+ break;
+ case STREAMTYPE_FILE_MONITOR:
+ CHKiRet(strmHandleEOFMonitor(pThis));
+ break;
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+/* read the next buffer from disk
+ * rgerhards, 2008-02-13
+ */
+static rsRetVal
+strmReadBuf(strm_t *pThis)
+{
+ DEFiRet;
+ int bRun;
+ long iLenRead;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+ /* We need to try read at least twice because we may run into EOF and need to switch files. */
+ bRun = 1;
+ while(bRun) {
+ /* first check if we need to (re)open the file. We may have switched to a new one in
+ * circular mode or it may have been rewritten (rotated) if we monitor a file
+ * rgerhards, 2008-02-13
+ */
+ CHKiRet(strmOpenFile(pThis));
+ iLenRead = read(pThis->fd, pThis->pIOBuf, pThis->sIOBufSize);
+ dbgoprint((obj_t*) pThis, "file %d read %ld bytes\n", pThis->fd, iLenRead);
+ if(iLenRead == 0) {
+ CHKiRet(strmHandleEOF(pThis));
+ } else if(iLenRead < 0)
+ ABORT_FINALIZE(RS_RET_IO_ERROR);
+ else { /* good read */
+ pThis->iBufPtrMax = iLenRead;
+ bRun = 0; /* exit loop */
+ }
+ }
+ /* if we reach this point, we had a good read */
+ pThis->iBufPtr = 0;
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* logically "read" a character from a file. What actually happens is that
+ * data is taken from the buffer. Only if the buffer is full, data is read
+ * directly from file. In that case, a read is performed blockwise.
+ * rgerhards, 2008-01-07
+ * NOTE: needs to be enhanced to support sticking with a strm entry (if not
+ * deleted).
+ */
+rsRetVal strmReadChar(strm_t *pThis, uchar *pC)
+{
+ DEFiRet;
+
+ ASSERT(pThis != NULL);
+ ASSERT(pC != NULL);
+
+ /* DEV debug only: dbgoprint((obj_t*) pThis, "strmRead index %d, max %d\n", pThis->iBufPtr, pThis->iBufPtrMax); */
+ if(pThis->iUngetC != -1) { /* do we have an "unread" char that we need to provide? */
+ *pC = pThis->iUngetC;
+ ++pThis->iCurrOffs; /* one more octet read */
+ pThis->iUngetC = -1;
+ ABORT_FINALIZE(RS_RET_OK);
+ }
+
+ /* do we need to obtain a new buffer? */
+ if(pThis->iBufPtr >= pThis->iBufPtrMax) {
+ CHKiRet(strmReadBuf(pThis));
+ }
+
+ /* if we reach this point, we have data available in the buffer */
+
+ *pC = pThis->pIOBuf[pThis->iBufPtr++];
+ ++pThis->iCurrOffs; /* one more octet read */
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* unget a single character just like ungetc(). As with that call, there is only a single
+ * character buffering capability.
+ * rgerhards, 2008-01-07
+ */
+rsRetVal strmUnreadChar(strm_t *pThis, uchar c)
+{
+ ASSERT(pThis != NULL);
+ ASSERT(pThis->iUngetC == -1);
+ pThis->iUngetC = c;
+ --pThis->iCurrOffs; /* one less octet read - NOTE: this can cause problems if we got a file change
+ and immediately do an unread and the file is on a buffer boundary and the stream is then persisted.
+ With the queue, this can not happen as an Unread is only done on record begin, which is never split
+ accross files. For other cases we accept the very remote risk. -- rgerhards, 2008-01-12 */
+
+ return RS_RET_OK;
+}
+
+
+/* read a line from a strm file. A line is terminated by LF. The LF is read, but it
+ * is not returned in the buffer (it is discared). The caller is responsible for
+ * destruction of the returned CStr object! -- rgerhards, 2008-01-07
+ * rgerhards, 2008-03-27: I now use the ppCStr directly, without any interim
+ * string pointer. The reason is that this function my be called by inputs, which
+ * are pthread_killed() upon termination. So if we use their native pointer, they
+ * can cleanup (but only then).
+ */
+rsRetVal
+strmReadLine(strm_t *pThis, cstr_t **ppCStr)
+{
+ DEFiRet;
+ uchar c;
+
+ ASSERT(pThis != NULL);
+ ASSERT(ppCStr != NULL);
+
+ CHKiRet(rsCStrConstruct(ppCStr));
+
+ /* now read the line */
+ CHKiRet(strmReadChar(pThis, &c));
+ while(c != '\n') {
+ CHKiRet(rsCStrAppendChar(*ppCStr, c));
+ CHKiRet(strmReadChar(pThis, &c));
+ }
+ CHKiRet(rsCStrFinish(*ppCStr));
+
+finalize_it:
+ if(iRet != RS_RET_OK && *ppCStr != NULL)
+ rsCStrDestruct(ppCStr);
+
+ RETiRet;
+}
+
+
+/* Standard-Constructor for the strm object
+ */
+BEGINobjConstruct(strm) /* be sure to specify the object type also in END macro! */
+ pThis->iCurrFNum = 1;
+ pThis->fd = -1;
+ pThis->iUngetC = -1;
+ pThis->sType = STREAMTYPE_FILE_SINGLE;
+ pThis->sIOBufSize = glblGetIOBufSize();
+ pThis->tOpenMode = 0600; /* TODO: make configurable */
+ENDobjConstruct(strm)
+
+
+/* ConstructionFinalizer
+ * rgerhards, 2008-01-09
+ */
+rsRetVal strmConstructFinalize(strm_t *pThis)
+{
+ DEFiRet;
+
+ ASSERT(pThis != NULL);
+
+ if(pThis->pIOBuf == NULL) { /* allocate our io buffer in case we have not yet */
+ if((pThis->pIOBuf = (uchar*) malloc(sizeof(uchar) * pThis->sIOBufSize)) == NULL)
+ ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+ pThis->iBufPtrMax = 0; /* results in immediate read request */
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* destructor for the strm object */
+BEGINobjDestruct(strm) /* be sure to specify the object type also in END and CODESTART macros! */
+CODESTARTobjDestruct(strm)
+ if(pThis->tOperationsMode == STREAMMODE_WRITE)
+ strmFlush(pThis);
+
+ /* ... then free resources */
+ if(pThis->fd != -1)
+ strmCloseFile(pThis);
+
+ if(pThis->pszDir != NULL)
+ free(pThis->pszDir);
+ if(pThis->pIOBuf != NULL)
+ free(pThis->pIOBuf);
+ if(pThis->pszCurrFName != NULL)
+ free(pThis->pszCurrFName);
+ if(pThis->pszFName != NULL)
+ free(pThis->pszFName);
+ENDobjDestruct(strm)
+
+
+/* check if we need to open a new file (in output mode only).
+ * The decision is based on file size AND record delimition state.
+ * This method may also be called on a closed file, in which case
+ * it immediately returns.
+ */
+static rsRetVal strmCheckNextOutputFile(strm_t *pThis)
+{
+ DEFiRet;
+
+ if(pThis->fd == -1)
+ FINALIZE;
+
+ if(pThis->iCurrOffs >= pThis->iMaxFileSize) {
+ dbgoprint((obj_t*) pThis, "max file size %ld reached for %d, now %ld - starting new file\n",
+ (long) pThis->iMaxFileSize, pThis->fd, (long) pThis->iCurrOffs);
+ CHKiRet(strmNextFile(pThis));
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+/* write memory buffer to a stream object.
+ * To support direct writes of large objects, this method may be called
+ * with a buffer pointing to some region other than the stream buffer itself.
+ * However, in that case the stream buffer must be empty (strmFlush() has to
+ * be called before), because we would otherwise mess up with the sequence
+ * inside the stream. -- rgerhards, 2008-01-10
+ */
+static rsRetVal strmWriteInternal(strm_t *pThis, uchar *pBuf, size_t lenBuf)
+{
+ DEFiRet;
+ int iWritten;
+
+ ASSERT(pThis != NULL);
+ ASSERT(pBuf == pThis->pIOBuf || pThis->iBufPtr == 0);
+
+ if(pThis->fd == -1)
+ CHKiRet(strmOpenFile(pThis));
+
+ iWritten = write(pThis->fd, pBuf, lenBuf);
+ dbgoprint((obj_t*) pThis, "file %d write wrote %d bytes\n", pThis->fd, iWritten);
+ /* TODO: handle error case -- rgerhards, 2008-01-07 */
+
+ /* Now indicate buffer empty again. We do this in any case, because there
+ * is no way we could react more intelligently to an error during write.
+ * This MUST be done BEFORE strCheckNextOutputFile(), otherwise we have an
+ * endless loop. We reset the buffer pointer also in finalize_it - this is
+ * necessary if we run into problems. Not resetting it would again cause an
+ * endless loop. So it is better to loose some data (which also justifies
+ * duplicating that code, too...) -- rgerhards, 2008-01-10
+ */
+ pThis->iBufPtr = 0;
+ pThis->iCurrOffs += iWritten;
+ /* update user counter, if provided */
+ if(pThis->pUsrWCntr != NULL)
+ *pThis->pUsrWCntr += iWritten;
+
+ if(pThis->sType == STREAMTYPE_FILE_CIRCULAR)
+ CHKiRet(strmCheckNextOutputFile(pThis));
+
+finalize_it:
+ pThis->iBufPtr = 0; /* see comment above */
+
+ RETiRet;
+}
+
+
+/* flush stream output buffer to persistent storage. This can be called at any time
+ * and is automatically called when the output buffer is full.
+ * rgerhards, 2008-01-10
+ */
+rsRetVal strmFlush(strm_t *pThis)
+{
+ DEFiRet;
+
+ ASSERT(pThis != NULL);
+ dbgoprint((obj_t*) pThis, "file %d flush, buflen %ld\n", pThis->fd, (long) pThis->iBufPtr);
+
+ if(pThis->tOperationsMode == STREAMMODE_WRITE && pThis->iBufPtr > 0) {
+ iRet = strmWriteInternal(pThis, pThis->pIOBuf, pThis->iBufPtr);
+ }
+
+ RETiRet;
+}
+
+
+/* seek a stream to a specific location. Pending writes are flushed, read data
+ * is invalidated.
+ * rgerhards, 2008-01-12
+ */
+static rsRetVal strmSeek(strm_t *pThis, off_t offs)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+
+ if(pThis->fd == -1)
+ strmOpenFile(pThis);
+ else
+ strmFlush(pThis);
+ int i;
+ dbgoprint((obj_t*) pThis, "file %d seek, pos %ld\n", pThis->fd, (long) offs);
+ i = lseek(pThis->fd, offs, SEEK_SET); // TODO: check error!
+ pThis->iCurrOffs = offs; /* we are now at *this* offset */
+ pThis->iBufPtr = 0; /* buffer invalidated */
+
+ RETiRet;
+}
+
+
+/* seek to current offset. This is primarily a helper to readjust the OS file
+ * pointer after a strm object has been deserialized.
+ */
+rsRetVal strmSeekCurrOffs(strm_t *pThis)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+
+ iRet = strmSeek(pThis, pThis->iCurrOffs);
+ RETiRet;
+}
+
+
+/* write a *single* character to a stream object -- rgerhards, 2008-01-10
+ */
+rsRetVal strmWriteChar(strm_t *pThis, uchar c)
+{
+ DEFiRet;
+
+ ASSERT(pThis != NULL);
+
+ /* if the buffer is full, we need to flush before we can write */
+ if(pThis->iBufPtr == pThis->sIOBufSize) {
+ CHKiRet(strmFlush(pThis));
+ }
+ /* we now always have space for one character, so we simply copy it */
+ *(pThis->pIOBuf + pThis->iBufPtr) = c;
+ pThis->iBufPtr++;
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* write an integer value (actually a long) to a stream object */
+rsRetVal strmWriteLong(strm_t *pThis, long i)
+{
+ DEFiRet;
+ uchar szBuf[32];
+
+ ASSERT(pThis != NULL);
+
+ CHKiRet(srUtilItoA((char*)szBuf, sizeof(szBuf), i));
+ CHKiRet(strmWrite(pThis, szBuf, strlen((char*)szBuf)));
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* write memory buffer to a stream object
+ */
+rsRetVal strmWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf)
+{
+ DEFiRet;
+ size_t iPartial;
+
+ ASSERT(pThis != NULL);
+ ASSERT(pBuf != NULL);
+
+ /* check if the to-be-written data is larger than our buffer size */
+ if(lenBuf >= pThis->sIOBufSize) {
+ /* it is - so we do a direct write, that is most efficient.
+ * TODO: is it really? think about disk block sizes!
+ */
+ CHKiRet(strmFlush(pThis)); /* we need to flush first!!! */
+ CHKiRet(strmWriteInternal(pThis, pBuf, lenBuf));
+ } else {
+ /* data fits into a buffer - we just need to see if it
+ * fits into the current buffer...
+ */
+ if(pThis->iBufPtr + lenBuf > pThis->sIOBufSize) {
+ /* nope, so we must split it */
+ iPartial = pThis->sIOBufSize - pThis->iBufPtr; /* this fits in current buf */
+ if(iPartial > 0) { /* the buffer was exactly full, can not write anything! */
+ memcpy(pThis->pIOBuf + pThis->iBufPtr, pBuf, iPartial);
+ pThis->iBufPtr += iPartial;
+ }
+ CHKiRet(strmFlush(pThis)); /* get a new buffer for rest of data */
+ memcpy(pThis->pIOBuf, pBuf + iPartial, lenBuf - iPartial);
+ pThis->iBufPtr = lenBuf - iPartial;
+ } else {
+ /* we have space, so we simply copy over the string */
+ memcpy(pThis->pIOBuf + pThis->iBufPtr, pBuf, lenBuf);
+ pThis->iBufPtr += lenBuf;
+ }
+ }
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* property set methods */
+/* simple ones first */
+DEFpropSetMeth(strm, bDeleteOnClose, int)
+DEFpropSetMeth(strm, iMaxFileSize, int)
+DEFpropSetMeth(strm, iFileNumDigits, int)
+DEFpropSetMeth(strm, tOperationsMode, int)
+DEFpropSetMeth(strm, tOpenMode, mode_t)
+DEFpropSetMeth(strm, sType, strmType_t);
+
+rsRetVal strmSetiMaxFiles(strm_t *pThis, int iNewVal)
+{
+ pThis->iMaxFiles = iNewVal;
+ pThis->iFileNumDigits = getNumberDigits(iNewVal);
+ return RS_RET_OK;
+}
+
+rsRetVal strmSetiAddtlOpenFlags(strm_t *pThis, int iNewVal)
+{
+ DEFiRet;
+
+ if(iNewVal & O_APPEND)
+ ABORT_FINALIZE(RS_RET_PARAM_ERROR);
+
+ pThis->iAddtlOpenFlags = iNewVal;
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* set the stream's file prefix
+ * The passed-in string is duplicated. So if the caller does not need
+ * it any longer, it must free it.
+ * rgerhards, 2008-01-09
+ */
+rsRetVal
+strmSetFName(strm_t *pThis, uchar *pszName, size_t iLenName)
+{
+ DEFiRet;
+
+ ASSERT(pThis != NULL);
+ ASSERT(pszName != NULL);
+
+ if(iLenName < 1)
+ ABORT_FINALIZE(RS_RET_FILE_PREFIX_MISSING);
+
+ if(pThis->pszFName != NULL)
+ free(pThis->pszFName);
+
+ if((pThis->pszFName = malloc(sizeof(uchar) * iLenName + 1)) == NULL)
+ ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+
+ memcpy(pThis->pszFName, pszName, iLenName + 1); /* always think about the \0! */
+ pThis->lenFName = iLenName;
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* set the stream's directory
+ * The passed-in string is duplicated. So if the caller does not need
+ * it any longer, it must free it.
+ * rgerhards, 2008-01-09
+ */
+rsRetVal
+strmSetDir(strm_t *pThis, uchar *pszDir, size_t iLenDir)
+{
+ DEFiRet;
+
+ ASSERT(pThis != NULL);
+ ASSERT(pszDir != NULL);
+
+ if(iLenDir < 1)
+ ABORT_FINALIZE(RS_RET_FILE_PREFIX_MISSING);
+
+ if((pThis->pszDir = malloc(sizeof(uchar) * iLenDir + 1)) == NULL)
+ ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
+
+ memcpy(pThis->pszDir, pszDir, iLenDir + 1); /* always think about the \0! */
+ pThis->lenDir = iLenDir;
+
+finalize_it:
+ RETiRet;
+}
+
+
+/* support for data records
+ * The stream class is able to write to multiple files. However, there are
+ * situation (actually quite common), where a single data record should not
+ * be split across files. This may be problematic if multiple stream write
+ * calls are used to create the record. To support that, we provide the
+ * bInRecord status variable. If it is set, no file spliting occurs. Once
+ * it is set to 0, a check is done if a split is necessary and it then
+ * happens. For a record-oriented caller, the proper sequence is:
+ *
+ * strmRecordBegin()
+ * strmWrite...()
+ * strmRecordEnd()
+ *
+ * Please note that records do not affect the writing of output buffers. They
+ * are always written when full. The only thing affected is circular files
+ * creation. So it is safe to write large records.
+ *
+ * IMPORTANT: RecordBegin() can not be nested! It is a programming error
+ * if RecordBegin() is called while already in a record!
+ *
+ * rgerhards, 2008-01-10
+ */
+rsRetVal strmRecordBegin(strm_t *pThis)
+{
+ ASSERT(pThis != NULL);
+ ASSERT(pThis->bInRecord == 0);
+ pThis->bInRecord = 1;
+ return RS_RET_OK;
+}
+
+rsRetVal strmRecordEnd(strm_t *pThis)
+{
+ DEFiRet;
+ ASSERT(pThis != NULL);
+ ASSERT(pThis->bInRecord == 1);
+
+ pThis->bInRecord = 0;
+ iRet = strmCheckNextOutputFile(pThis); /* check if we need to switch files */
+
+ RETiRet;
+}
+/* end stream record support functions */
+
+
+/* This method serializes a stream object. That means the whole
+ * object is modified into text form. That text form is suitable for
+ * later reconstruction of the object.
+ * The most common use case for this method is the creation of an
+ * on-disk representation of the message object.
+ * We do not serialize the dynamic properties.
+ * rgerhards, 2008-01-10
+ */
+rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm)
+{
+ DEFiRet;
+ int i;
+ long l;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+ ISOBJ_TYPE_assert(pStrm, strm);
+
+ strmFlush(pThis);
+ CHKiRet(obj.BeginSerialize(pStrm, (obj_t*) pThis));
+
+ objSerializeSCALAR(pStrm, iCurrFNum, INT);
+ objSerializePTR(pStrm, pszFName, PSZ);
+ objSerializeSCALAR(pStrm, iMaxFiles, INT);
+ objSerializeSCALAR(pStrm, bDeleteOnClose, INT);
+
+ i = pThis->sType;
+ objSerializeSCALAR_VAR(pStrm, sType, INT, i);
+
+ i = pThis->tOperationsMode;
+ objSerializeSCALAR_VAR(pStrm, tOperationsMode, INT, i);
+
+ i = pThis->tOpenMode;
+ objSerializeSCALAR_VAR(pStrm, tOpenMode, INT, i);
+
+ l = (long) pThis->iCurrOffs;
+ objSerializeSCALAR_VAR(pStrm, iCurrOffs, LONG, l);
+
+ CHKiRet(obj.EndSerialize(pStrm));
+
+finalize_it:
+ RETiRet;
+}
+
+
+
+/* set a user write-counter. This counter is initialized to zero and
+ * receives the number of bytes written. It is accurate only after a
+ * flush(). This hook is provided as a means to control disk size usage.
+ * The pointer must be valid at all times (so if it is on the stack, be sure
+ * to remove it when you exit the function). Pointers are removed by
+ * calling strmSetWCntr() with a NULL param. Only one pointer is settable,
+ * any new set overwrites the previous one.
+ * rgerhards, 2008-02-27
+ */
+rsRetVal
+strmSetWCntr(strm_t *pThis, number_t *pWCnt)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+
+ if(pWCnt != NULL)
+ *pWCnt = 0;
+ pThis->pUsrWCntr = pWCnt;
+
+ RETiRet;
+}
+
+
+#include "stringbuf.h"
+
+/* This function can be used as a generic way to set properties.
+ * rgerhards, 2008-01-11
+ */
+#define isProp(name) !rsCStrSzStrCmp(pProp->pcsName, (uchar*) name, sizeof(name) - 1)
+rsRetVal strmSetProperty(strm_t *pThis, var_t *pProp)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+ ASSERT(pProp != NULL);
+
+ if(isProp("sType")) {
+ CHKiRet(strmSetsType(pThis, (strmType_t) pProp->val.num));
+ } else if(isProp("iCurrFNum")) {
+ pThis->iCurrFNum = pProp->val.num;
+ } else if(isProp("pszFName")) {
+ CHKiRet(strmSetFName(pThis, rsCStrGetSzStrNoNULL(pProp->val.pStr), rsCStrLen(pProp->val.pStr)));
+ } else if(isProp("tOperationsMode")) {
+ CHKiRet(strmSettOperationsMode(pThis, pProp->val.num));
+ } else if(isProp("tOpenMode")) {
+ CHKiRet(strmSettOpenMode(pThis, pProp->val.num));
+ } else if(isProp("iCurrOffs")) {
+ pThis->iCurrOffs = pProp->val.num;
+ } else if(isProp("iMaxFileSize")) {
+ CHKiRet(strmSetiMaxFileSize(pThis, pProp->val.num));
+ } else if(isProp("iMaxFiles")) {
+ CHKiRet(strmSetiMaxFiles(pThis, pProp->val.num));
+ } else if(isProp("iFileNumDigits")) {
+ CHKiRet(strmSetiFileNumDigits(pThis, pProp->val.num));
+ } else if(isProp("bDeleteOnClose")) {
+ CHKiRet(strmSetbDeleteOnClose(pThis, pProp->val.num));
+ }
+
+finalize_it:
+ RETiRet;
+}
+#undef isProp
+
+
+/* return the current offset inside the stream. Note that on two consequtive calls, the offset
+ * reported on the second call may actually be lower than on the first call. This is due to
+ * file circulation. A caller must deal with that. -- rgerhards, 2008-01-30
+ */
+rsRetVal
+strmGetCurrOffset(strm_t *pThis, int64 *pOffs)
+{
+ DEFiRet;
+
+ ISOBJ_TYPE_assert(pThis, strm);
+ ASSERT(pOffs != NULL);
+
+ *pOffs = pThis->iCurrOffs;
+
+ RETiRet;
+}
+
+
+/* queryInterface function
+ * rgerhards, 2008-02-29
+ */
+BEGINobjQueryInterface(strm)
+CODESTARTobjQueryInterface(strm)
+ if(pIf->ifVersion != strmCURR_IF_VERSION) { /* check for current version, increment on each change */
+ ABORT_FINALIZE(RS_RET_INTERFACE_NOT_SUPPORTED);
+ }
+
+ /* ok, we have the right interface, so let's fill it
+ * Please note that we may also do some backwards-compatibility
+ * work here (if we can support an older interface version - that,
+ * of course, also affects the "if" above).
+ */
+ //xxxpIf->oID = OBJvm;
+
+finalize_it:
+ENDobjQueryInterface(strm)
+
+
+/* Initialize the stream class. Must be called as the very first method
+ * before anything else is called inside this class.
+ * rgerhards, 2008-01-09
+ */
+BEGINObjClassInit(strm, 1, OBJ_IS_CORE_MODULE)
+ /* request objects we use */
+
+ OBJSetMethodHandler(objMethod_SERIALIZE, strmSerialize);
+ OBJSetMethodHandler(objMethod_SETPROPERTY, strmSetProperty);
+ OBJSetMethodHandler(objMethod_CONSTRUCTION_FINALIZER, strmConstructFinalize);
+ENDObjClassInit(strm)
+
+
+/*
+ * vi:set ai:
+ */