/*
 * Initialization for the stream opened by the other process
 */
static int
openStream(Stream *stream)
{
    jint error;

    CHECK_ERROR(sysIPMutexOpen(stream->shared->mutexName, &stream->mutex));

    error = sysEventOpen(stream->shared->hasDataEventName,
                             &stream->hasData);
    if (error != SYS_OK) {
        setLastError(error);
        (void)closeStream(stream, JNI_FALSE);
        return error;
    }

    error = sysEventOpen(stream->shared->hasSpaceEventName,
                             &stream->hasSpace);
    if (error != SYS_OK) {
        setLastError(error);
        (void)closeStream(stream, JNI_FALSE);
        return error;
    }

    stream->state = STATE_OPEN;

    return SYS_OK;
}

/*
 * Initialization for the stream opened by the other process
 */
static int
openStream(Stream *stream)
{
    jint error;

    error = sysIPMutexOpen(stream->shared->mutexName, &stream->mutex);
    if (error != SYS_OK) {
        fprintf(stderr,"Error accessing mutex, rc = %d\n", error);
        return error;
    }

    error = sysEventOpen(stream->shared->hasDataEventName,
                             &stream->hasData);
    if (error != SYS_OK) {
        fprintf(stderr,"Error accessing mutex, rc = %d\n", error);
        (void)closeStream(stream);
        return error;
    }

    error = sysEventOpen(stream->shared->hasSpaceEventName,
                             &stream->hasSpace);
    if (error != SYS_OK) {
        fprintf(stderr,"Error accessing mutex, rc = %d\n", error);
        (void)closeStream(stream);
        return error;
    }

    stream->state = STATE_OPEN;

    return SYS_OK;
}

/*
 * Server creates stream.
 */
static int
createStream(char *name, Stream *stream)
{
    jint error;
    char prefix[MAX_IPC_PREFIX];

    sprintf(prefix, "%s.mutex", name);
    error = createWithGeneratedName(prefix, stream->shared->mutexName,
                                    createMutex, &stream->mutex);
    if (error != SYS_OK) {
        return error;
    }

    sprintf(prefix, "%s.hasData", name);
    error = createWithGeneratedName(prefix, stream->shared->hasDataEventName,
                                    createEvent, &stream->hasData);
    if (error != SYS_OK) {
        (void)closeStream(stream, JNI_FALSE);
        return error;
    }

    sprintf(prefix, "%s.hasSpace", name);
    error = createWithGeneratedName(prefix, stream->shared->hasSpaceEventName,
                                    createEvent, &stream->hasSpace);
    if (error != SYS_OK) {
        (void)closeStream(stream, JNI_FALSE);
        return error;
    }

    stream->shared->readOffset = 0;
    stream->shared->writeOffset = 0;
    stream->shared->isFull = JNI_FALSE;
    stream->state = STATE_OPEN;
    return SYS_OK;
}

static void
closeConnection(SharedMemoryConnection *connection)
{
    /*
     * Signal all threads accessing this connection that we are
     * shutting down.
     */
    if (connection->shutdown) {
	sysEventSignal(connection->shutdown);
    }


    (void)closeStream(&connection->incoming);
    (void)closeStream(&connection->outgoing);

    if (connection->sharedMemory) {
	sysSharedMemClose(connection->sharedMemory, connection->shared);
    }
    if (connection->otherProcess) {
	sysProcessClose(connection->otherProcess);
    }

    /*
     * Ideally we should close the connection->shutdown event and 
     * free the connection structure. However as closing the 
     * connection is asynchronous it means that other threads may
     * still be accessing the connection structure. On Win32 this
     * means we leak 132 bytes and one event per connection. This
     * memory will be reclaim at process exit.
     *
     * if (connection->shutdown) 
     *     sysEventClose(connection->shutdown);
     * freeConnection(connection);
     */
}

bool AlsaCtlPortConfig::sendToHW(string &error)
{
    PortConfig portConfig;
    blackboardRead(&portConfig, sizeof(portConfig));

    // If device update is needed, close all the stream
    if (isDeviceUpdateNeeded(portConfig)) {

        // Close playback and capture
        closeStream(Playback);
        closeStream(Capture);

        // Save new configuration
        _portConfig.channelNumber = portConfig.channelNumber;
        _portConfig.format = portConfig.format;
        _portConfig.sampleRate = portConfig.sampleRate;

    } else {

        // Close playback/capture streams if asked for
        if (!portConfig.isStreamEnabled[Playback]) {

            closeStream(Playback);
        }
        if (!portConfig.isStreamEnabled[Capture]) {

            closeStream(Capture);
        }
    }

    // Open and configure required streams
    if (portConfig.isStreamEnabled[Playback]) {

        if (!openStream(Playback, error)) {

            return false;
        }
    }
    if (portConfig.isStreamEnabled[Capture]) {

        if (!openStream(Capture, error)) {

            return false;
        }
    }

    // Check port configuration has been considered
    assert(!memcmp(&_portConfig, &portConfig, sizeof(_portConfig)));

    return true;
}

I8KView::~I8KView()
{
  closeStream();

  delete m_timer;
  delete m_reData;
}

void test_cftNew_should_create_cft_for_symbols_from_file(){
  Stream *in = NULL;
  CFT *cftPtr;
  int tableSize;
  CEXCEPTION_T error;
  
  Try{
  // Please check cftData.txt, inside has a lengthy string 
    in = openStream("cftData.txt","r");
    cftPtr = cftNew(in, &tableSize);
  }Catch(error){
    TEST_ASSERT_NOT_EQUAL(ERR_FILE_NOT_EXIST,error);
  }
  dumpCFT(cftPtr,0);
  dumpCFT(cftPtr,1);
  dumpCFT(cftPtr,2);
  
  TEST_ASSERT_EQUAL(3,tableSize);
  
  TEST_ASSERT_EQUAL('a', cftPtr[0].symbol);
  TEST_ASSERT_EQUAL(40, cftPtr[0].cum_Freq);
  TEST_ASSERT_EQUAL(40, cftPtr[0].occurNo);

  TEST_ASSERT_EQUAL('b', cftPtr[1].symbol);
  TEST_ASSERT_EQUAL(41, cftPtr[1].cum_Freq);
  TEST_ASSERT_EQUAL(1, cftPtr[1].occurNo);

  TEST_ASSERT_EQUAL('c', cftPtr[2].symbol);
  TEST_ASSERT_EQUAL(50, cftPtr[2].cum_Freq);
  TEST_ASSERT_EQUAL(9, cftPtr[2].occurNo);
  
  if(in != NULL){
    closeStream(in);
  }
}

LadybugWidget::~LadybugWidget()
{
  // make sure the thread gets stopped and stream closed
  closeStream();

  delete mNextFrame;
}

void plotly::reconnectStream(){
    while(!client.connected()){
        if(log_level<4) Serial.println(F("... Disconnected from streaming servers"));
        closeStream();
        openStream();
    }
}

FILE*
HTTPClient::openClientFile()
{
  FILE* result = fdevopen(clientWrite, clientRead);
  if (result == NULL)
    {
      return NULL;
    }
  http_stream_udata* udata = (http_stream_udata*) malloc(
      sizeof(http_stream_udata));
  fdev_set_udata(result,udata);
  udata->client = this;
  udata->encode = 0;
  if (connected())
    {
      stop();
    }
  if (connect())
    {
      return result;
    }
  else
    {
      closeStream(result);
      return NULL;
    }
}

audio::orchestra::api::Core::~Core() {
	// The subclass destructor gets called before the base class
	// destructor, so close an existing stream before deallocating
	// apiDeviceId memory.
	if (m_state != audio::orchestra::state::closed) {
		closeStream();
	}
}

void PacketReceiver::exit()
{
	destroyFrames();
	destroyPictureConverter();
	closeStream();
	mainWindow->log("Stream geschlossen", INFO);
	exitAVLibs();
}

JavaInputStream::~JavaInputStream() {
	JNIEnv *env = AndroidUtil::getEnv();
	if (myJavaInputStream != 0) {
		closeStream(env);
	}
	env->DeleteGlobalRef(myJavaFile);
	env->DeleteGlobalRef(myJavaBuffer);
}

void handleClientRequest(int i)
{
    ssize_t r;
    char * p;
    size_t remain;

    if (stream[i].len >= sizeof(stream[i].buffer) - 2)
    {
        logAbort("Input line on stream %d too long: %s\n", i, stream[i].buffer);
    }
    remain = sizeof(stream[i].buffer) - stream[i].len - 2;

    logDebug("handleClientRequest: read i=%d\n", i);
    logDebug("read %s i=%d fd=%d len=%u remain=%u\n", streamTypeName[stream[i].type], i, stream[i].fd, stream[i].len, remain);
    r = read(stream[i].fd, stream[i].buffer + stream[i].len, remain);

    if (r <= 0)
    {
        logDebug("read %s i=%d fd=%d r=%d\n", streamTypeName[stream[i].type], i, stream[i].fd, r);
        if (stream[i].type == DATA_INPUT_STREAM)
        {
            logAbort("EOF on reading stdin\n");
        }
        closeStream(i);
        return;
    }

    stream[i].len += r;
    stream[i].buffer[stream[i].len] = 0;
    while (stream[i].len > 0)
    {
        size_t len;

        p = strchr(stream[i].buffer, '\n');
        if (!p)
        {
            break;
        }
        len = p - stream[i].buffer;
        sbAppendData(&tcpMessage, stream[i].buffer, len + 1);
        if (stream[i].type != DATA_INPUT_STREAM || stream[outputIdx].type == DATA_OUTPUT_COPY)
        {
            /* Send all TCP client input and the main stdin stream if the mode is -o */
            /* directly to stdout */
            sbAppendData(&outMessage, stream[i].buffer, len + 1);
        }
        *p = 0;
        if (storeMessage(stream[i].buffer, len))
        {
            convertJSONToNMEA0183(&nmeaMessage, stream[i].buffer);
        }
        p++, len++;
        stream[i].len -= len;

        /* Now remove [buffer..p> == the entire line */
        memmove(stream[i].buffer, p, stream[i].len + 1);
    }
}

 void NodeDataInterface::streamToOutput(AbstractWriteFacet &write_to_storeentity) {
         static char readBuffer[4096];
         size_t bytesRead;
         openStream(); 
         while((bytesRead = streamRead(readBuffer, 4096)) > 0){
              write_to_storeentity(readBuffer, bytesRead);
         }
         closeStream();
 }

// destructor
slgAudio::~slgAudio(){
    if( m_audio )
    {
        stopStream();
        closeStream();
    }
    delete m_audio;
    m_audio = NULL;
}

	//! Disconnect old streams
	void SimpleDashelConnection::closeOldStreams()
	{
		for (size_t i = 0; i < toDisconnect.size(); ++i)
		{
			SEND_NOTIFICATION(LOG_WARNING, "old client disconnected", toDisconnect[i]->getTargetName());
			closeStream(toDisconnect[i]);
		}
		toDisconnect.clear();
	}

void ConsoleAppender::close()
{
  QMutexLocker locker(&mObjectGuard);

  if (isClosed())
    return;

  WriterAppender::close();
  closeStream();
}

int NetworkInputStream::checkStatus() {
    if (_open) {
        int res = waitAvailable(2000);
        if (res < 0) {
            closeStream();
        }
        return res;
    }
    return 0;
}

void StreamSubscriptionResponderBase::onNextFrame(Frame_CANCEL&& frame) {
  switch (state_) {
    case State::RESPONDING:
      state_ = State::CLOSED;
      closeStream(StreamCompletionSignal::CANCEL);
      break;
    case State::CLOSED:
      break;
  }
}