@Override
public void persist(PolicyDAO policy) {
    PolicyDAOImpl impl = getImpl(policy, PolicyDAOImpl.class);
    UUID policyID = impl.getPolicyID().getUUID();

    Mutator<UUID> m = Schema.POLICIES.createMutator(_keyspace);

    Schema.POLICIES.SHORT_NAME.addInsertion(m, policyID, policy.getShortName());
    Schema.POLICIES.DESCRIPTION.addInsertion(m, policyID, policy.getDescription());

    // We're saving changes, so update the edit time
    Schema.POLICIES.LAST_EDITED.addInsertion(m, policyID, new Date());

    // TODO: error handling? Throws HectorException.
    m.execute();
}
 

@Override
public void incrementAggregateCounters( UUID userId, UUID groupId, String category, Map<String, Long> counters ) {

    // TODO shortcircuit
    if ( !skipAggregateCounters ) {
        long timestamp = cass.createTimestamp();
        Mutator<ByteBuffer> m = createMutator( cass.getApplicationKeyspace( applicationId ), be );
        counterUtils.batchIncrementAggregateCounters(
            m, applicationId, userId, groupId, null, category, counters, timestamp );

        //Adding graphite metrics
        Timer.Context timeIncrementCounters =entIncrementAggregateCountersTimer.time();
        CassandraPersistenceUtils.batchExecute( m, CassandraService.RETRY_COUNT );
        timeIncrementCounters.stop();
    }
}
 

/**
 * Flushes the buffer.
 * 
 * @param mutator The mutator to fill with deletions.
 * @param columnFamilyName The column family (name) to delete from.
 * @param column The column to delete from.
 * @param serializer The serializer to use for the column.
 * @param dao the rdf index data access object.
 * @param <T> The type of the column key.
 * @return True if the buffer was flushed and at least one element was checked for deletion, false otherwise.
 * @throws DataAccessLayerException in case of data access failure.
 */
<T> boolean flush(
		final Mutator<byte[]> mutator, 
		final String columnFamilyName, 
		final T column, 
		final Serializer<T> serializer,
		final TripleIndexDAO dao) throws DataAccessLayerException {
	if (_candidates.size() == 0) {
		return false;
	}
	
	for (SecondaryIndexDeletionCandidate candidate : _candidates) {
		if (!dao.query(candidate.getQuery(), 1).hasNext()) {
			mutator.addDeletion(candidate.getRow(), columnFamilyName, column, serializer);
		}
	}
	
	return true;
}
 

private void handleAggregateCounterRow( Mutator<ByteBuffer> m, String key, long column, long value,
                                        UUID applicationId ) {
    if ( logger.isTraceEnabled() ) {
        logger.trace( "HACR: aggregateRow for app {} with key {} column {} and value {}",
                applicationId, key, column, value );
    }
    if ( "o".equals( counterType ) || "p".equals( counterType ) ) {
        if ( m != null ) {
            HCounterColumn<Long> c = createCounterColumn( column, value, le );
            m.addCounter( bytebuffer( key ), APPLICATION_AGGREGATE_COUNTERS.toString(), c );
        }
    }
    if ( "n".equals( counterType ) || "p".equals( counterType ) ) {
        // create and add Count
        PrefixedSerializer ps =
                new PrefixedSerializer( applicationId, ue, se );
        batcher.add(
                new Count( APPLICATION_AGGREGATE_COUNTERS.toString(), ps.toByteBuffer( key ), column, value ) );
    }
}
 

/**
 * Adds a role to the role store.
 */
@Override
public void doAddRole(String roleName, String[] userList, boolean shared) throws UserStoreException {

    Mutator<Composite> mutator = HFactory.createMutator(keyspace, CompositeSerializer.get());
    Composite composite = new Composite();
    composite.addComponent(roleName, stringSerializer);
    composite.addComponent(tenantIdString, stringSerializer);

    mutator.addInsertion(composite, CFConstants.UM_ROLES,
            HFactory.createColumn(CFConstants.UM_ROLE_NAME, roleName, stringSerializer, stringSerializer));
    mutator.addInsertion(composite, CFConstants.UM_ROLES,
            HFactory.createColumn(CFConstants.UM_TENANT_ID, tenantIdString, stringSerializer, stringSerializer));

    if (userList != null && userList.length > 0) {
        addRoleToUsersList(userList, roleName, mutator);
    }

    mutator.execute();
}
 

/**
 * Maps the users to a role list. Adds the (username, tenantId) -> roleList
 * and (role, tenantId) -> userName
 *
 * @param userName The username of the user the roles need to be added to.
 * @param roleList The list of roles that needs to be mapped against the user.
 */
private void addUserToRoleList(String userName, String[] roleList) {

    Mutator<Composite> mutator = HFactory.createMutator(keyspace, CompositeSerializer.get());

    if (roleList != null) {
        for (String role : roleList) {
            Composite key = new Composite();
            key.addComponent(userName, stringSerializer);
            key.addComponent(tenantIdString, stringSerializer);

            mutator.addInsertion(key, CFConstants.UM_USER_ROLE, HFactory.createColumn(role, role));

            Composite keyRole = new Composite();
            keyRole.addComponent(role, stringSerializer);
            keyRole.addComponent(tenantIdString, stringSerializer);

            mutator.addInsertion(keyRole, CFConstants.UM_ROLE_USER_INDEX, HFactory.createColumn(userName, userName));

        }
        mutator.execute();
    }
}
 

/**
 * Maps the users to a role list. Adds the (username, tenantId) -> roleList
 * and (role, tenantId) -> userName
 *
 * @param userName The username of the user the roles need to be added to.
 * @param roleList The list of roles that needs to be mapped against the user.
 * @param mutator  Passes the mutator and returns it with the insert statements.
 */
private Mutator<Composite> addUserToRoleList(String userName, String[] roleList, Mutator<Composite> mutator) {
    if (roleList != null && mutator != null) {
        for (String role : roleList) {
            Composite key = new Composite();
            key.addComponent(userName, stringSerializer);
            key.addComponent(tenantIdString, stringSerializer);

            mutator.addInsertion(key, CFConstants.UM_USER_ROLE, HFactory.createColumn(role, role));

            Composite keyRole = new Composite();
            keyRole.addComponent(role, stringSerializer);
            keyRole.addComponent(tenantIdString, stringSerializer);

            mutator.addInsertion(keyRole, CFConstants.UM_ROLE_USER_INDEX, HFactory.createColumn(userName, userName));

        }
    }
    return mutator;
}
 

protected void recordLicenses( String projectVersionMetadataKey, List<License> licenses )
{

    if ( licenses == null || licenses.isEmpty() )
    {
        return;
    }
    Mutator<String> licenseMutator = this.licenseTemplate.createMutator();

    for ( License license : licenses )
    {
        // we don't care about the key as the real used one with the projectVersionMetadata
        String keyLicense = UUID.randomUUID().toString();
        String cfLicense = cassandraArchivaManager.getLicenseFamilyName();

        addInsertion( licenseMutator, keyLicense, cfLicense, "projectVersionMetadataModel.key",
                      projectVersionMetadataKey );

        addInsertion( licenseMutator, keyLicense, cfLicense, NAME.toString(), license.getName() );

        addInsertion( licenseMutator, keyLicense, cfLicense, URL.toString(), license.getUrl() );

    }
    licenseMutator.execute();
}
 

public static void addInsertToMutator( Mutator<ByteBuffer> m, Object columnFamily, Object key, Object columnName,
                                       Object columnValue, long timestamp ) {

    logBatchOperation( "Insert", columnFamily, key, columnName, columnValue, timestamp );

    if ( columnName instanceof List<?> ) {
        columnName = DynamicComposite.toByteBuffer( ( List<?> ) columnName );
    }
    if ( columnValue instanceof List<?> ) {
        columnValue = DynamicComposite.toByteBuffer( ( List<?> ) columnValue );
    }

    HColumn<ByteBuffer, ByteBuffer> column =
            createColumn( bytebuffer( columnName ), bytebuffer( columnValue ), timestamp, be, be );
    m.addInsertion( bytebuffer( key ), columnFamily.toString(), column );
}
 

public void batchIncrementAggregateCounters( Mutator<ByteBuffer> m, UUID applicationId, UUID userId, UUID groupId,
                                             UUID queueId, String category, String name, long value,
                                             long counterTimestamp, long cassandraTimestamp ) {
    for ( CounterResolution resolution : CounterResolution.values() ) {
        if (logger.isTraceEnabled()) {
            logger.trace("BIAC for resolution {}", resolution);
        }

        batchIncrementAggregateCounters( m, userId, groupId, queueId, category, resolution, name, value,
                counterTimestamp, applicationId );

        if (logger.isTraceEnabled()) {
            logger.trace("DONE BIAC for resolution {}", resolution);
        }
    }
    batchIncrementEntityCounter( m, applicationId, name, value, cassandraTimestamp, applicationId );
    if ( userId != null ) {
        batchIncrementEntityCounter( m, userId, name, value, cassandraTimestamp, applicationId );
    }
    if ( groupId != null ) {
        batchIncrementEntityCounter( m, groupId, name, value, cassandraTimestamp, applicationId );
    }
}
 

public void batchUpdateQueuePropertiesIndexes( Mutator<ByteBuffer> batch, UUID publisherQueueId,
                                               String subscriberQueuePath, UUID subscriberQueueId,
                                               Map<String, Object> properties, UUID timestampUuid )
        throws Exception {

    for ( Map.Entry<String, Object> property : properties.entrySet() ) {

        if ( !Queue.QUEUE_PROPERTIES.containsKey( property.getKey() ) ) {

            QueueIndexUpdate indexUpdate =
                    batchStartQueueIndexUpdate( batch, subscriberQueuePath, subscriberQueueId, property.getKey(),
                            property.getValue(), timestampUuid );

            batchUpdateQueueIndex( indexUpdate, publisherQueueId );
        }
    }
}
 

/**
 * Write the updated client pointer
 *
 * @param lastReturnedId This is a null safe parameter. If it's null, this won't be written since it means we didn't
 * read any messages
 */
protected void writeClientPointer( UUID queueId, UUID consumerId, UUID lastReturnedId ) {
    // nothing to do
    if ( lastReturnedId == null ) {
        return;
    }

    // we want to set the timestamp to the value from the time uuid. If this is
    // not the max time uuid to ever be written
    // for this consumer, we want this to be discarded to avoid internode race
    // conditions with clock drift.
    long colTimestamp = UUIDUtils.getTimestampInMicros( lastReturnedId );

    Mutator<UUID> mutator = CountingMutator.createFlushingMutator( ko, ue );

    if ( logger.isDebugEnabled() ) {
        logger.debug( "Writing last client id pointer of '{}' for queue '{}' and consumer '{}' with timestamp '{}",
                        lastReturnedId, queueId, consumerId, colTimestamp
                );
    }

    mutator.addInsertion( consumerId, CONSUMERS.getColumnFamily(),
            createColumn( queueId, lastReturnedId, colTimestamp, ue, ue ) );

    mutator.execute();
}
 

@Override
public void addSetToDictionary( EntityRef entityRef, String dictionaryName, Set<?> elementValues )
        throws Exception {

    if ( ( elementValues == null ) || elementValues.isEmpty() ) {
        return;
    }

    EntityRef entity = get( entityRef );

    UUID timestampUuid = UUIDUtils.newTimeUUID();
    Mutator<ByteBuffer> batch = createMutator( cass.getApplicationKeyspace( applicationId ), be );

    for ( Object elementValue : elementValues ) {
        batch = batchUpdateDictionary( batch, entity, dictionaryName, elementValue, null, false, timestampUuid );
    }

    //Adding graphite metrics
    Timer.Context timeAddingSetDictionary = entAddDictionarySetTimer.time();
    CassandraPersistenceUtils.batchExecute( batch, CassandraService.RETRY_COUNT );
    timeAddingSetDictionary.stop();
}
 

public Message storeEventAsMessage(Mutator<ByteBuffer> m, Event event, long timestamp) {

        counterUtils.addEventCounterMutations(m, applicationId, event, timestamp);

        QueueManager q = queueManagerFactory.getQueueManager(applicationId);

        Message message = new Message();
        message.setType("event");
        message.setCategory(event.getCategory());
        message.setStringProperty("message", event.getMessage());
        message.setTimestamp(timestamp);
        q.postToQueue("events", message);

        return message;
    }
 

public void insert(Mutator<String> mutator, String key,
                   Map<String, String> columnsMap, String columnFamily,
                   String superColumn) {
    List<HColumn<String, String>> columns = new ArrayList<HColumn<String, String>>();
    for (String name : columnsMap.keySet()) {
        HFactory.createStringColumn(name, columnsMap.get(name));
    }
    mutator.insert(key, columnFamily, HFactory.createSuperColumn(superColumn, columns, stringSerializer, stringSerializer, stringSerializer));
}
 

@Override
public void revokeGroupRolePermission( UUID groupId, String roleName, String permission ) throws Exception {
    roleName = roleName.toLowerCase();
    permission = permission.toLowerCase();
    long timestamp = cass.createTimestamp();
    Mutator<ByteBuffer> batch = createMutator( cass.getApplicationKeyspace( applicationId ), be );
    CassandraPersistenceUtils.addDeleteToMutator(batch, ApplicationCF.ENTITY_DICTIONARIES,
        getRolePermissionsKey(groupId, roleName), permission, timestamp);
    //Adding graphite metrics
    Timer.Context timeRevokeGroupRolePermission = entRevokeGroupPermissionTimer.time();
    CassandraPersistenceUtils.batchExecute( batch, CassandraService.RETRY_COUNT );
    timeRevokeGroupRolePermission.stop();
}
 

@Override
public void setAll(final Map<K, V> pairs) {
	final Mutator<K> mutator = createMutator(_keyspace, _serializer_k);

	for (final K key : pairs.keySet()) {
		mutator.addInsertion(key, _cf_name, createColumn(COLUMN_NAME, pairs.get(key), BYTE_SERIALIZER, _serializer_v));
	}

	try {
		mutator.execute();
	} catch (final Exception exception) {
		_log.error(MessageCatalog._00057_ADD_FAILURE, exception);
	}
}
 

@Override
public Mutator<ByteBuffer> batchSetProperty( Mutator<ByteBuffer> batch, EntityRef entity,
        String propertyName, Object propertyValue, boolean force, boolean noRead,
        UUID timestampUuid ) throws Exception {

    throw new UnsupportedOperationException( "Not supported yet." );
}
 

@Override
public void doUpdateCredentialByAdmin(String userName, Object newCredential) throws UserStoreException {
    if (!checkUserPasswordValid(newCredential)) {
        throw new UserStoreException(
                "Credential not valid. Credential must be a non null string with following format, "
                        + realmConfig.getUserStoreProperty(UserCoreConstants.RealmConfig.PROPERTY_JAVA_REG_EX));

    }

    String saltValue = null;
    if (TRUE.equalsIgnoreCase(realmConfig.getUserStoreProperties().get(JDBCRealmConstants.STORE_SALTED_PASSWORDS))) {
        saltValue = Util.getSaltValue();
    }
    String password = Util.preparePassword((String) newCredential, saltValue);
    Mutator<Composite> mutator = HFactory.createMutator(keyspace, CompositeSerializer.get());
    Composite key = new Composite();
    key.addComponent(userName, stringSerializer);
    key.addComponent(tenantIdString, stringSerializer);
    mutator.addInsertion(key, CFConstants.UM_USER,
            HFactory.createColumn(CFConstants.UM_SECRET, password, stringSerializer, stringSerializer));
    mutator.addInsertion(key, CFConstants.UM_USER,
            HFactory.createColumn(CFConstants.UM_SALT_VALUE, saltValue, stringSerializer, stringSerializer));
    try {
        mutator.execute();
        if (log.isDebugEnabled()) {
            log.debug("Changed password for user " + userName + "successfully");
        }
    } catch (HectorException e) {
        throw new UserStoreException("Change Password failed.", e);
    }
}
 

public void addMessageCounterMutations( Mutator<ByteBuffer> m, UUID applicationId, UUID queueId, Message msg,
                                        long timestamp ) {
    if ( msg.getCounters() != null ) {
        for ( Entry<String, Integer> value : msg.getCounters().entrySet() ) {
            batchIncrementAggregateCounters( m, applicationId, null, null, queueId, msg.getCategory(),
                    value.getKey().toLowerCase(), value.getValue(), msg.getTimestamp(), timestamp );
        }
    }
}
 

private Mutator<Composite> addClaimsForUser(String userName, Map<String, String> claims, Mutator<Composite> mutator) {

        Composite key = new Composite();
        key.addComponent(userName, stringSerializer);
        key.addComponent(tenantIdString, stringSerializer);
        // add claims
        for (Map.Entry<String, String> claimsVals : claims.entrySet()) {
            mutator.addInsertion(key, CFConstants.UM_USER_ATTRIBUTE,
                    HFactory.createColumn(claimsVals.getKey(), claimsVals.getValue()));
            mutator.addInsertion(key, CFConstants.UM_USER_ATTRIBUTE,
                    HFactory.createColumn(CFConstants.UM_TENANT_ID, tenantIdString));
        }
        return mutator;
    }
 

public void batchIncrementAggregateCounters( Mutator<ByteBuffer> m, UUID applicationId, UUID userId, UUID groupId,
                                             UUID queueId, String category, Map<String, Long> counters,
                                             long timestamp ) {
    if ( counters != null ) {
        for ( Entry<String, Long> value : counters.entrySet() ) {
            batchIncrementAggregateCounters( m, applicationId, userId, groupId, queueId, category,
                    value.getKey().toLowerCase(), value.getValue(), timestamp );
        }
    }
}
 

public static void addInsertion( Mutator<String> mutator, String key, String columnFamily, String columnName,
                                 String value )
{
    if ( value != null )
    {
        mutator.addInsertion( key, columnFamily, column( columnName, value ) );
    }
}
 

@Override
public void grantGroupRolePermission( UUID groupId, String roleName, String permission ) throws Exception {
    roleName = roleName.toLowerCase();
    permission = permission.toLowerCase();
    long timestamp = cass.createTimestamp();
    Mutator<ByteBuffer> batch = createMutator( cass.getApplicationKeyspace( applicationId ), be );
    CassandraPersistenceUtils.addInsertToMutator(batch, ApplicationCF.ENTITY_DICTIONARIES,
        getRolePermissionsKey(groupId, roleName), permission, ByteBuffer.allocate(0), timestamp);

    //Adding graphite metrics
    Timer.Context timeGroupRolePermission = entGrantGroupPermissionTimer.time();
    CassandraPersistenceUtils.batchExecute(batch, CassandraService.RETRY_COUNT);
    timeGroupRolePermission.stop();
}
 

protected void recordMailingList( String projectVersionMetadataKey, List<MailingList> mailingLists )
{
    if ( mailingLists == null || mailingLists.isEmpty() )
    {
        return;
    }
    Mutator<String> mailingMutator = this.mailingListTemplate.createMutator();
    for ( MailingList mailingList : mailingLists )
    {
        // we don't care about the key as the real used one with the projectVersionMetadata
        String keyMailingList = UUID.randomUUID().toString();
        String cfMailingList = cassandraArchivaManager.getMailingListFamilyName();

        addInsertion( mailingMutator, keyMailingList, cfMailingList, "projectVersionMetadataModel.key",
                      projectVersionMetadataKey );
        addInsertion( mailingMutator, keyMailingList, cfMailingList, NAME.toString(), mailingList.getName() );
        addInsertion( mailingMutator, keyMailingList, cfMailingList, "mainArchiveUrl",
                      mailingList.getMainArchiveUrl() );
        addInsertion( mailingMutator, keyMailingList, cfMailingList, "postAddress", mailingList.getPostAddress() );
        addInsertion( mailingMutator, keyMailingList, cfMailingList, "subscribeAddress",
                      mailingList.getSubscribeAddress() );
        addInsertion( mailingMutator, keyMailingList, cfMailingList, "unsubscribeAddress",
                      mailingList.getUnsubscribeAddress() );
        int idx = 0;
        for ( String otherArchive : mailingList.getOtherArchives() )
        {
            addInsertion( mailingMutator, keyMailingList, cfMailingList, "otherArchive." + idx, otherArchive );
            idx++;
        }

    }
    mailingMutator.execute();
}
 

public void setColumn( Keyspace ko, Object columnFamily, Object key, Object columnName, Object columnValue,
                       int ttl ) throws Exception {

    if ( db_logger.isTraceEnabled() ) {
        db_logger.trace( "setColumn cf={} key={} name={} value={}", columnFamily, key, columnName, columnValue );
    }

    ByteBuffer name_bytes = null;
    if ( columnName instanceof List ) {
        name_bytes = DynamicComposite.toByteBuffer( ( List<?> ) columnName );
    }
    else {
        name_bytes = bytebuffer( columnName );
    }

    ByteBuffer value_bytes = null;
    if ( columnValue instanceof List ) {
        value_bytes = DynamicComposite.toByteBuffer( ( List<?> ) columnValue );
    }
    else {
        value_bytes = bytebuffer( columnValue );
    }

    HColumn<ByteBuffer, ByteBuffer> col = createColumn( name_bytes, value_bytes, be, be );
    if ( ttl != 0 ) {
        col.setTtl( ttl );
    }
    Mutator<ByteBuffer> m = CountingMutator.createFlushingMutator( ko, be );
    m.insert( bytebuffer( key ), columnFamily.toString(), col );
}
 

public Mutator<ByteBuffer> batchIncrementQueueCounters( Mutator<ByteBuffer> m, Map<UUID, Map<String, Long>> values,
                                                        long timestamp, UUID applicationId ) {
    for ( Entry<UUID, Map<String, Long>> entry : values.entrySet() ) {
        batchIncrementQueueCounters( m, entry.getKey(), entry.getValue(), timestamp, applicationId );
    }
    return m;
}
 

@Override
public Message postToQueue( String queuePath, Message message ) {
    long timestamp = cass.createTimestamp();
    Mutator<ByteBuffer> batch = CountingMutator.createFlushingMutator( cass.getApplicationKeyspace( applicationId ),
            be );

    queuePath = normalizeQueuePath( queuePath );

    MessageIndexUpdate indexUpdate = new MessageIndexUpdate( message );

    batchPostToQueue( batch, queuePath, message, indexUpdate, timestamp );

    batchExecute( batch, RETRY_COUNT );

    String firstSubscriberQueuePath = null;
    while ( true ) {

        QueueSet subscribers = getSubscribers( queuePath, firstSubscriberQueuePath, 1000 );

        if ( subscribers.getQueues().isEmpty() ) {
            break;
        }

        batch = CountingMutator.createFlushingMutator( cass.getApplicationKeyspace( applicationId ), be );
        for ( QueueInfo q : subscribers.getQueues() ) {
            batchPostToQueue( batch, q.getPath(), message, indexUpdate, timestamp );

            firstSubscriberQueuePath = q.getPath();
        }
        batchExecute( batch, RETRY_COUNT );

        if ( !subscribers.hasMore() ) {
            break;
        }
    }

    return message;
}
 

public void batchSubscribeToQueue( Mutator<ByteBuffer> batch, String publisherQueuePath, UUID publisherQueueId,
                                   String subscriberQueuePath, UUID subscriberQueueId, long timestamp ) {

    batch.addInsertion( bytebuffer( publisherQueueId ), QUEUE_SUBSCRIBERS.getColumnFamily(),
            createColumn( subscriberQueuePath, subscriberQueueId, timestamp, se, ue ) );

    batch.addInsertion( bytebuffer( subscriberQueueId ), QUEUE_SUBSCRIPTIONS.getColumnFamily(),
            createColumn( publisherQueuePath, publisherQueueId, timestamp, se, ue ) );
}
 

@Override
public QueueSet subscribeToQueue( String publisherQueuePath, String subscriberQueuePath ) {

    publisherQueuePath = normalizeQueuePath( publisherQueuePath );
    UUID publisherQueueId = getQueueId( publisherQueuePath );

    subscriberQueuePath = normalizeQueuePath( subscriberQueuePath );
    UUID subscriberQueueId = getQueueId( subscriberQueuePath );

    UUID timestampUuid = newTimeUUID();
    long timestamp = getTimestampInMicros( timestampUuid );

    Mutator<ByteBuffer> batch = CountingMutator.createFlushingMutator( cass.getApplicationKeyspace( applicationId ),
            be );

    batchSubscribeToQueue( batch, publisherQueuePath, publisherQueueId, subscriberQueuePath, subscriberQueueId,
            timestamp );

    try {
        Queue queue = getQueue( subscriberQueuePath, subscriberQueueId );
        if ( queue != null ) {
            batchUpdateQueuePropertiesIndexes( batch, publisherQueueId, subscriberQueuePath, subscriberQueueId,
                    queue.getProperties(), timestampUuid );
        }
    }
    catch ( Exception e ) {
        logger.error( "Unable to update index", e );
    }

    batchExecute( batch, RETRY_COUNT );

    return new QueueSet().addQueue( subscriberQueuePath, subscriberQueueId );
}