docs: public void reconstitutePubSubSourceStreams(StreamSet streamSet, int startMode)
                      throws
                      SIErrorException, SIDiscriminatorSyntaxException, SIResourceException
      {
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.entry(tc, "reconstitutePubSubSourceStreams", 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 = _txManager.createAutoCommitTransaction();
              try
              {
                  streamSet.requestUpdate(tran);
              } catch (MessageStoreException e)
              {
                  // MessageStoreException shouldn't occur so FFDC.
                  FFDCFilter.processException(
                                              e,
                                              "com.ibm.ws.sib.processor.impl.PubSubInputHandler.reconstitutePubSubSourceStreams",
                                              "1:3145:1.329.1.1",
                                              this);
  
                  SibTr.exception(tc, e);
                  if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                      SibTr.exit(tc, "reconstitutePubSubSourceStreams", "SIStoreException");
                  throw new SIResourceException(e);
              }
          }
  
          try
          {
              NonLockingCursor cursor = _proxyReferenceStream.newNonLockingCursor(
                              new ClassEqualsFilter(MessageItemReference.class));
              cursor.allowUnavailableItems();
              MessageItemReference ref = (MessageItemReference) cursor.next();
  
              // Rebuild internalOutputStreams by running refs through matchspace 
              LocalTransaction siTran = null;
              Transaction msTran = null;
              int batchCount = 0;
              while (ref != null)
              {
                  // If the count of reconstituted messages has reached 50, commit the transaction
                  if (batchCount >= 50) {
                      try
                      {
                          siTran.commit();
                          // Null out the transaction object so that a new one is created next
                          siTran = null;
                      } catch (SIIncorrectCallException e)
                      {
                          // No FFDC code needed
                          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                              SibTr.exit(tc, "reconstitutePubSubSourceStreams", "SIResourceException");
                          throw new SIResourceException(e);
                      }
                  }
  
                  // If the transaction object is null then we've either never created it or have just committed it.
                  // Create a new transaction 
                  if (siTran == null) {
                      siTran = _txManager.createLocalTransaction(true);
                      msTran = (Transaction) siTran;
                      batchCount = 0;
                  }
  
                  boolean commitInsert = false;
  
                  // Change streamID in message to streamID of StreamSet
                  // If we are restoring from a stale backup this will
                  // ensure that the restoreFanOut method puts it on a new InteralOutputStream
                  if (ref.getGuaranteedStreamUuid() != streamSet.getStreamID())
                  {
                      ref.setGuaranteedStreamUuid(streamSet.getStreamID());
                  }
  
                  if (!(ref.isAdding() || ref.isRemoving()))
                      commitInsert = true;
                  // If there are no matching neighbours, remove reference
                  if (!restoreFanOut(ref, commitInsert))
                  {
                      if (!(ref.isAdding() || ref.isRemoving()))
                      {
                          ref.remove(msTran, ref.getLockID());
                          batchCount++;
                      }
                      else
                      // TODO: ref is indoubt, need to mark it for later deletion on commit
                      {
                      }
                  }
                  ref = (MessageItemReference) cursor.next();
              }
              try
              {
                  if (siTran != null) {
                      siTran.commit();
                  }
              } catch (SIIncorrectCallException e)
              {
                  // No FFDC code needed
                  if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                      SibTr.exit(tc, "reconstitutePubSubSourceStreams", "SIResourceException");
                  throw new SIResourceException(e);
              }
  
              // Ditch the cursor & create again to get back to head of list
              cursor.finished();
              cursor = _proxyReferenceStream.newNonLockingCursor(
                              new ClassEqualsFilter(MessageItemReference.class));
              cursor.allowUnavailableItems();
              ref = (MessageItemReference) cursor.next();
  
              // Add all messages back in to the SourceStreams
              while (ref != 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 (ref.getGuaranteedStreamUuid() != streamSet.getStreamID())
                  {
                      ref.setGuaranteedStreamUuid(streamSet.getStreamID());
                  }
  
                  //add all messages back in to the streams
                  if (!(ref.isAdding() || ref.isRemoving()))
                  {
                      // commit those which are not in doubt
                      _sourceStreamManager.restoreMessage(ref, true);
                  }
                  else
                  {
                      // add to stream in uncommitted state
                      _sourceStreamManager.restoreMessage(ref, false);
                  }
  
                  ref = (MessageItemReference) cursor.next();
              }
              // Consolidate all streams which may have been reconstituted
              _sourceStreamManager.consolidateStreams(startMode);
  
              cursor.finished();
  
          } catch (MessageStoreException e)
          {
              // MessageStoreException shouldn't occur so FFDC.
              FFDCFilter.processException(
                                          e,
                                          "com.ibm.ws.sib.processor.impl.PubSubInputHandler.reconstitutePubSubSourceStreams",
                                          "1:3277:1.329.1.1",
                                          this);
  
              SibTr.exception(tc, e);
              SibTr.error(tc, "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                          new Object[] {
                                        "com.ibm.ws.sib.processor.impl.PubSubInputHandler",
                                        "1:3284:1.329.1.1",
                                        e });
  
              if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                  SibTr.exit(tc, "reconstitutePubSubSourceStreams", e);
  
              throw new SIResourceException(
                              nls.getFormattedMessage(
                                                      "INTERNAL_MESSAGING_ERROR_CWSIP0002",
                                                      new Object[] {
                                                                    "com.ibm.ws.sib.processor.impl.PubSubInputHandler",
                                                                    "1:3295:1.329.1.1",
                                                                    e },
                                                      null),
                              e);
          }
  
          if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
              SibTr.exit(tc, "reconstitutePubSubSourceStreams");
      }