Create a new Source Stream and store it in the given StreamSet

@param streamSet
@param priority
@param reliability
@return a new SourceStream
@throws SIResourceException
975003
private SourceStream createStream(StreamSet streamSet,
                              int priority,
                              Reliability reliability) throws SIResourceException

  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "createStream", new Object[] {streamSet, new Integer(priority), reliability});

    SourceStream stream = null;

    //there is no source stream for express messages
    if(reliability.compareTo(Reliability.BEST_EFFORT_NONPERSISTENT) > 0)
    {
      //Warning - this assumes that ASSURED is always the highest Reliability
      //and that UNKNOWN is always the lowest (0).
      stream = new SourceStream(priority, //priority
                        reliability, //reliability
                        downControl,
                        new ArrayList(),
                        streamSet,
                        messageProcessor.getAlarmManager(),
                        destinationHandler);
    }

    streamSet.setStream(priority, reliability, stream);

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "createStream", stream);

    return stream;
  }
1239149
@Task(name = "createStream", version = "1.0", resource = "{scope}/{stream}")
    public CompletableFuture<CreateStreamStatus.Status> createStream(String scope, String stream, StreamConfiguration config, long createTimestamp) {
        return execute(
                new Resource(scope, stream),
                new Serializable[]{scope, stream, config, createTimestamp},
                () -> createStreamBody(scope, stream, config, createTimestamp));
    }
496707
public <U> BaseSimpleReactStream<U> fromStream(final Stream<CompletableFuture<U>> stream) {

        final Stream s = stream;
        return construct(s);
    }
62962
public void reconstituteStreamSet(StreamSet streamSet)
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(this, tc, "reconstituteStreamSet", streamSet);

    synchronized(streamSets)
    {
      streamSets.put(streamSet.getStreamID(), streamSet);
      sourceMap.put(streamSet.getStreamID(), streamSet.getRemoteMEUuid());
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "reconstituteStreamSet");
  }
262040
public void processAckExpected(ControlAckExpected ackExpMsg) throws SIResourceException 
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "processAckExpectedMessage", new Object[] { ackExpMsg });
    
    int priority = ackExpMsg.getPriority().intValue();
    Reliability reliability = ackExpMsg.getReliability();
    
    SIBUuid12 streamID = ackExpMsg.getGuaranteedStreamUUID();
    StreamSet streamSet = getStreamSet(streamID, true);
  
    InternalInputStream internalInputStream = null;
    synchronized(streamSet)
    {
      internalInputStream = (InternalInputStream) streamSet.getStream(priority, reliability);        

      // This may be the first message which has required this stream    
      if(internalInputStream == null && 
      (reliability.compareTo(Reliability.BEST_EFFORT_NONPERSISTENT) > 0))
      {
        internalInputStream = createStream(streamSet, priority, reliability);      
      }
    }
        
    internalInputStream.processAckExpected(ackExpMsg.getTick());
        
    if (tc.isEntryEnabled())
      SibTr.exit(tc, "processAckExpectedMessage");
  }
222544
private function _createCurrentStream($priorityName, $subPriority = null)
    {
        $priorityName = strtolower($priorityName);

        if ($subPriority !== null) {
            $priorityName = strtolower($subPriority);
        }

        $streamDir = $this->baseDir . DIRECTORY_SEPARATOR . str_ireplace('-', DIRECTORY_SEPARATOR, $this->date->format('Y-m-d'));

        if (!is_dir($streamDir)) {
            umask(0000);
            if (!mkdir($streamDir, 0777, true)) {
                $msg = "Dir '$streamDir' cannot be created";
                throw new \Zend_Log_Exception($msg);
            }

            $this->shutdown($priorityName);
        }

        $filepath = $streamDir . DIRECTORY_SEPARATOR . $priorityName . '.log';
        if (!$this->streams[$priorityName] = @fopen($filepath, $this->mode, false)) {
            $msg = "File '$filepath' cannot be opened with mode '$this->mode'";
            throw new \Zend_Log_Exception($msg);
        }

        return $this->streams[$priorityName];
    }
405852
public CompletableFuture<UpdateStreamStatus.Status> updateStream(String scope, String stream, StreamConfiguration newConfig,
                                                                     OperationContext contextOpt) {
        final OperationContext context = contextOpt == null ? streamMetadataStore.createContext(scope, stream) : contextOpt;
        final long requestId = requestTracker.getRequestIdFor("updateStream", scope, stream);

        // 1. get configuration
        return streamMetadataStore.getConfigurationRecord(scope, stream, context, executor)
                .thenCompose(configProperty -> {
                    // 2. post event to start update workflow
                    if (!configProperty.getObject().isUpdating()) {
                        return addIndexAndSubmitTask(new UpdateStreamEvent(scope, stream, requestId), 
                                // 3. update new configuration in the store with updating flag = true
                                // if attempt to update fails, we bail out with no harm done
                                () -> streamMetadataStore.startUpdateConfiguration(scope, stream, newConfig,
                                        context, executor))
                                // 4. wait for update to complete
                                .thenCompose(x -> checkDone(() -> isUpdated(scope, stream, newConfig, context))
                                        .thenApply(y -> UpdateStreamStatus.Status.SUCCESS));
                    } else {
                        log.warn(requestId, "Another update in progress for {}/{}",
                                scope, stream);
                        return CompletableFuture.completedFuture(UpdateStreamStatus.Status.FAILURE);
                    }
                })
                .exceptionally(ex -> {
                    log.warn(requestId, "Exception thrown in trying to update stream configuration {}",
                            ex.getMessage());
                    return handleUpdateStreamError(ex, requestId);
                });
    }
532963
public Reliability getReliability()
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.entry(tc, "getReliability");
    Reliability rel =  _sourceStream.getReliability();
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) 
      SibTr.exit(tc, "getReliability", rel);    
    return rel;
  }
1160143
@SuppressWarnings("unchecked")
    public static <T> ConnectableFlowable<T> createFrom(Flowable<? extends T> source) {
        Flowable<T> sourceCast = (Flowable<T>) source;
        return create(sourceCast, DEFAULT_UNBOUNDED_FACTORY);
    }
83952
public static Builder newBuilder(TableId sourceTable, List<String> destinationUris) {
    return new Builder().setSourceTable(sourceTable).setDestinationUris(destinationUris);
  }
1304332
private <OUT> DataStreamSource<OUT> fromParallelCollection(SplittableIterator<OUT> iterator, TypeInformation<OUT>
			typeInfo, String operatorName) {
		return addSource(new FromSplittableIteratorFunction<>(iterator), operatorName, typeInfo);
	}
951506
public boolean addMessage(SIMPMessage msgItem) throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "addMessage", new Object[] { msgItem });
  
    // Get the JsMessage as we need to update the Guaranteed fields
    JsMessage jsMsg = msgItem.getMessage();
  
    StreamSet streamSet = getStreamSet();
  
    // Stamp the message with the stream ID
    msgItem.setGuaranteedStreamUuid(streamSet.getStreamID());
  
    //  NOTE: no synchronization with flush necessary here since startFlush
    // is not supposed to be called until all competing producer threads have
    // exited.
    Reliability reliability = msgItem.getReliability();
  
    if(reliability == Reliability.BEST_EFFORT_NONPERSISTENT)
    {
      addBestEffortMessage(msgItem);
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "addMessage", false);
      return false;
    }
    else
    {
      SourceStream sourceStream = null;
      int priority = msgItem.getPriority();
      synchronized(streamSet)
      {        
        sourceStream = (SourceStream) streamSet.getStream(priority, reliability);
    
        if(sourceStream == null)
        {
          sourceStream = createStream(streamSet,
                                      priority,
                                      reliability,
                                      streamSet.getPersistentData(priority, reliability),
                                      false );
        }
      }
    
      synchronized(sourceStream)
      {    
        long tick = this.messageProcessor.nextTick();
    
        //Set a unique id in the message if explicitly told to or
        //if one has not already been set
        if(msgItem.getRequiresNewId()  ||  jsMsg.getSystemMessageId()==null)
        {
          jsMsg.setSystemMessageSourceUuid(this.messageProcessor.getMessagingEngineUuid());
          jsMsg.setSystemMessageValue(tick);
          msgItem.setRequiresNewId(false);
        }
    
        jsMsg.setGuaranteedValueEndTick(tick);
        jsMsg.setGuaranteedValueValueTick(tick);
        jsMsg.setGuaranteedValueRequestedOnly(false);
    
        jsMsg.setGuaranteedValueStartTick(sourceStream.getLastMsgAdded() + 1);
        jsMsg.setGuaranteedValueCompletedPrefix(sourceStream.getCompletedPrefix());
    
        sourceStream.writeUncommitted(msgItem);
      } // end sync
      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "addMessage", true);
    
      return true;
    }    
  }
79469
private void processAckExpected(ControlAckExpected ackExpMsg)
                    throws SIResourceException
    {

        // This is called by a PubSubOutputHandler when it finds Unknown
        // ticks in it's own stream and need the InputStream to resend them
        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.entry(tc, "processAckExpected", ackExpMsg);

        // The upstream cellule for this stream is stored in the originStreamMap
        // unless we were the originator.
        SIBUuid12 streamID = ackExpMsg.getGuaranteedStreamUUID();
        int priority = ackExpMsg.getPriority().intValue();
        Reliability reliability = ackExpMsg.getReliability();

        if (_internalInputStreamManager.hasStream(streamID, priority, reliability))
        {
            _internalInputStreamManager.processAckExpected(ackExpMsg);
        }
        else
        {
            _targetStreamManager.handleAckExpectedMessage(ackExpMsg);
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
            SibTr.exit(tc, "processAckExpected");

    }
642612
public static InputStreamResource inputStreamResource(final String path,
                                                          final SupplierWithThrowable<InputStream, IOException> supplier) {
        return new InputStreamResource(path, supplier);
    }
1038038
public void reconstitutePtoPSourceStreams(StreamSet streamSet, int startMode ) 
  throws SIRollbackException, SIConnectionLostException, SIResourceException, SIErrorException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "reconstitutePtoPSourceStreams", new Object[] {streamSet, new Integer(startMode)});

    sourceStreamManager.reconstituteStreamSet(streamSet);
    
    // Don't do flush if we are asked to start in recovery mode 
    if(   ((startMode & JsConstants.ME_START_FLUSH ) == JsConstants.ME_START_FLUSH )
       && ((startMode & JsConstants.ME_START_RECOVERY ) == 0 ) )
   {
      this.sendFlushedMessage(null, streamSet.getStreamID());
      // Now change streamID of streamSet
      streamSet.setStreamID(new SIBUuid12());
      // This calls requestUpdate on the StreamSet Item which will
      // cause a callback to the streamSet.getPersistentData() by msgstore
      Transaction tran = messageProcessor.getTXManager().createAutoCommitTransaction();
      try
      {
        streamSet.requestUpdate(tran);
      }
      catch (MessageStoreException e)
      {
         // MessageStoreException shouldn't occur so FFDC.
         FFDCFilter.processException(
         e,
         "com.ibm.ws.sib.processor.impl.PtoPOutputHandler.reconstitutePtoPSourceStreams",
         "1:1883:1.241",
         this);
  
         SibTr.exception(tc, e);
         if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) SibTr.exit(tc, "reconstitutePtoPSourceStreams", "SIStoreException");
         throw new SIResourceException(e);
      }      
    }
        
    NonLockingCursor cursor = null;
    try
    {
      cursor = transmissionItemStream.newNonLockingItemCursor(
        new ClassEqualsFilter(MessageItem.class));
      cursor.allowUnavailableItems();  

      MessageItem msg = (MessageItem)cursor.next();
      while(msg!=null)
      {
        // Change streamID in message to streamID of StreamSet
        // If we are restoring from a stale backup this will
        // and the restoreMessage method puts it on a new SourceStream
        if( msg.getGuaranteedStreamUuid() !=  streamSet.getStreamID())
        {
          msg.setGuaranteedStreamUuid(streamSet.getStreamID());
        }
              
        //add all messages back in to the streams
        if(!(msg.isAdding() || msg.isRemoving()))
        {
          // commit those which are not in doubt
          sourceStreamManager.restoreMessage(msg, true);
        }
        else
        {
          // add to stream in uncommitted state
          sourceStreamManager.restoreMessage(msg, false);
        }
        
        msg = (MessageItem)cursor.next();
      }
      
      // Consolidate all streams which may have been reconstituted
      // This is necessary as messages may have been added out of order
      // This will return a list of the messageStoreIds of all the
      // messages which need to be locked because they are inside the
      // sendWindows of the streams which they were added to and so may
      // already have been sent 
      List msgsToLock = sourceStreamManager.consolidateStreams(startMode);
      
      // Run through locking all messages in list so they won't expire.      
      // watch out as list is Longs msgstore ids
      long msgId = 0;
      MessageItem msgItem = null;
      for (int i = 0; i < msgsToLock.size(); i++)
      {
        msgId = ((Long)msgsToLock.get(i)).longValue();
        msgItem = (MessageItem)transmissionItemStream.findById(msgId); 
        if( msgItem != null )
          msgItem.lockItemIfAvailable(lockID);
        else
        {
          if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            SibTr.debug(tc, "message wasn't there when we tried to lock it");
        }
      }    
    }
    catch (MessageStoreException e)
    {
      // MessageStoreException shouldn't occur so FFDC.
      FFDCFilter.processException(
        e,
        "com.ibm.ws.sib.processor.impl.PtoPOutputHandler.reconstitutePtoPSourceStreams",
        "1:1956:1.241",
        this);

      SibTr.exception(tc, e);
      SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
        new Object[] {
          "com.ibm.ws.sib.processor.impl.PtoPOutputHandler",
          "1:1963:1.241",
          e });

      if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
        SibTr.exit(tc, "reconstitutePtoPSourceStreams", e);

      throw new SIResourceException(
        nls.getFormattedMessage(
          "INTERNAL_MESSAGING_ERROR_CWSIP0002",
          new Object[] {
            "com.ibm.ws.sib.processor.impl.PtoPOutputHandler",
            "1:1974:1.241",
            e },
          null),
        e);
    }
    finally
    {
      if (cursor != null)
        cursor.finished();
    }

    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())  
      SibTr.exit(tc, "reconstitutePtoPSourceStreams");      
  }
1095021
public CounterManagerConfigurationBuilder reliability(Reliability reliability) {
      attributes.attribute(CounterManagerConfiguration.RELIABILITY).set(reliability);
      return this;
   }
434106
void processAckExpected(
    long ackExpStamp,
    int priority,
    Reliability reliability,
    SIBUuid12 stream)
   throws SIResourceException
  {
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.entry(tc, "processAckExpected", new Long(ackExpStamp));

    _internalOutputStreamManager.processAckExpected(ackExpStamp, priority, reliability, stream);
    
    if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
      SibTr.exit(tc, "processAckExpected");
  }