text:
package com.hurence.logisland.processor.datastore;

import com.hurence.logisland.classloading.PluginProxy;
import com.hurence.logisland.component.InitializationException;
import com.hurence.logisland.record.FieldType;
import com.hurence.logisland.record.Record;
import com.hurence.logisland.record.StandardRecord;
import com.hurence.logisland.service.datastore.DatastoreClientService;
import com.hurence.logisland.util.runner.TestRunner;
import com.hurence.logisland.util.runner.TestRunners;
import org.junit.Test;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;

import static org.junit.Assert.assertEquals;

public class EnrichRecordsTest {

    @Test
    public void testSimpleEnrichment() throws InitializationException {

        // create the controller service and link it to the test processor
        final DatastoreClientService service = new MockDatastoreService();

        final TestRunner runner = TestRunners.newTestRunner(new EnrichRecords());
        runner.setProperty(EnrichRecords.RECORD_KEY_FIELD, "record_id");
        runner.setProperty(EnrichRecords.COLLECTION_NAME, "test");
        runner.setProperty(EnrichRecords.DATASTORE_CLIENT_SERVICE, service.getIdentifier());
        runner.addControllerService(service.getIdentifier(), service);
        runner.enableControllerService(service);

        getLookupRecords().forEach(r -> service.put("test", r, false));

        final DatastoreClientService lookupService = PluginProxy.unwrap(runner.getProcessContext()
                .getPropertyValue(EnrichRecords.DATASTORE_CLIENT_SERVICE)
                .asControllerService());

        Collection<Record> recordsToEnrich = getRecords();

        runner.assertValid();
        runner.enqueue(recordsToEnrich);
        runner.run();
        runner.assertAllInputRecordsProcessed();
        runner.assertOutputRecordsCount(3);
        runner.assertOutputErrorCount(0);

        Record expected = getJoinedRecords().get("id1");
        Record enriched0 = runner.getOutputRecords().get(0);
        assertEquals(expected.getField("a"), enriched0.getField("a"));
        assertEquals(expected.getField("b"), enriched0.getField("b"));
        assertEquals(expected.getField("c"), enriched0.getField("c"));
        assertEquals(expected.getField("f1"), enriched0.getField("f1"));
        assertEquals(expected.getField("f2"), enriched0.getField("f2"));

    }

    private Collection<Record> getRecords() {
        Collection<Record> recordsToEnrich = new ArrayList<>();

        recordsToEnrich.add(new StandardRecord()
                .setId("id1")
                .setField("a", FieldType.STRING, "a1")
                .setField("b", FieldType.STRING, "b1")
                .setField("c", FieldType.LONG, 1));

        recordsToEnrich.add(new StandardRecord()
                .setId("id2")
                .setField("a", FieldType.STRING, "a2")
                .setField("b", FieldType.STRING, "b2")
                .setField("c", FieldType.LONG, 2));

        recordsToEnrich.add(new StandardRecord()
                .setId("id3")
                .setField("a", FieldType.STRING, "a3")
                .setField("b", FieldType.STRING, "b3")
                .setField("c", FieldType.LONG, 3));
        return recordsToEnrich;
    }

    private Map<String, Record> getJoinedRecords() {
        Map<String, Record> lookupRecords = new HashMap<>();

        lookupRecords.put("id1", new StandardRecord()
                .setId("id1")
                .setField("f1", FieldType.STRING, "value1")
                .setField("f2", FieldType.STRING, "falue2")
                .setField("a", FieldType.STRING, "a1")
                .setField("b", FieldType.STRING, "b1")
                .setField("c", FieldType.LONG, 1));

        lookupRecords.put("id2", new StandardRecord()
                .setId("id2")
                .setField("f1", FieldType.STRING, "value3")
                .setField("f2", FieldType.STRING, "falue4")
                .setField("a", FieldType.STRING, "a2")
                .setField("b", FieldType.STRING, "b2")
                .setField("c", FieldType.LONG, 2));

        lookupRecords.put("id3", new StandardRecord()
                .setId("id3")
                .setField("f1", FieldType.STRING, "value5")
                .setField("f2", FieldType.STRING, "falue6")
                .setField("a", FieldType.STRING, "a3")
                .setField("b", FieldType.STRING, "b3")
                .setField("c", FieldType.LONG, 3));

        return lookupRecords;
    }

    private Collection<Record> getLookupRecords() {
        Collection<Record> lookupRecords = new ArrayList<>();

        lookupRecords.add(new StandardRecord()
                .setId("id1")
                .setField("f1", FieldType.STRING, "value1")
                .setField("f2", FieldType.STRING, "falue2"));

        lookupRecords.add(new StandardRecord()
                .setId("id2")
                .setField("f1", FieldType.STRING, "value3")
                .setField("f2", FieldType.STRING, "falue4"));

        lookupRecords.add(new StandardRecord()
                .setId("id3")
                .setField("f1", FieldType.STRING, "value5")
                .setField("f2", FieldType.STRING, "falue6"));

        return lookupRecords;
    }
}


text:
package org.eclipse.jnosql.diana.keyvalue.query;

import jakarta.nosql.Params;
import jakarta.nosql.QueryException;
import jakarta.nosql.Value;
import jakarta.nosql.query.ArrayQueryValue;
import jakarta.nosql.query.Function;
import jakarta.nosql.query.FunctionQueryValue;
import jakarta.nosql.query.JSONQueryValue;
import jakarta.nosql.query.ParamQueryValue;
import jakarta.nosql.query.QueryValue;
import jakarta.nosql.query.ValueType;

import java.util.Arrays;
import java.util.stream.Stream;

import static java.util.stream.Collectors.toList;

final class Values {

    private Values() {
    }

    private static Object get(QueryValue<?> value, Params parameters) {

        ValueType type = value.getType();
        switch (type) {
            case NUMBER:
            case STRING:
                return value.get();
            case PARAMETER:
                return parameters.add(ParamQueryValue.class.cast(value).get());
            case ARRAY:
                return Stream.of(ArrayQueryValue.class.cast(value).get())
                        .map(v -> get(v, parameters))
                        .collect(toList());
            case FUNCTION:
                Function function = FunctionQueryValue.class.cast(value).get();
                String name = function.getName();
                Object[] params = function.getParams();
                if ("convert".equals(name)) {
                    return executeConvert(parameters, params);
                }
                String message = String.format("There is not support to the function: %s with parameters %s", name,
                        Arrays.toString(params));
                throw new QueryException(message);
            case JSON:
                return JSONQueryValue.class.cast(value).get().toString();
            case CONDITION:
            default:
                throw new QueryException("There is not support to the value: " + type);

        }
    }

    private static Object executeConvert(Params parameters, Object[] params) {
        Object value = get(QueryValue.class.cast(params[0]), parameters);
        return Value.of(value)
                .get((Class<?>) params[1]);
    }

    static Value getValue(QueryValue<?> value, Params parameters) {
        Object result = get(value, parameters);
        if (result instanceof Value) {
            return Value.class.cast(result);
        }
        return Value.of(result);
    }
}

redpajama_set_name: RedPajamaGithub


text:
using System;

namespace Lucene.Net.Index
{

    using DocValuesConsumer = Lucene.Net.Codecs.DocValuesConsumer;
    using Similarity = Lucene.Net.Search.Similarities.Similarity;

    internal sealed class NormsConsumerPerField : InvertedDocEndConsumerPerField, IComparable<NormsConsumerPerField>
    {
        private readonly FieldInfo FieldInfo;
        private readonly DocumentsWriterPerThread.DocState DocState;
        private readonly Similarity Similarity;
        private readonly FieldInvertState FieldState;
        private NumericDocValuesWriter Consumer;

        public NormsConsumerPerField(DocInverterPerField docInverterPerField, FieldInfo fieldInfo, NormsConsumer parent)
        {
            this.FieldInfo = fieldInfo;
            DocState = docInverterPerField.DocState;
            FieldState = docInverterPerField.FieldState;
            Similarity = DocState.Similarity;
        }

        public int CompareTo(NormsConsumerPerField other)
        {
            return FieldInfo.Name.CompareTo(other.FieldInfo.Name);
        }

        internal override void Finish()
        {
            if (FieldInfo.Indexed && !FieldInfo.OmitsNorms())
            {
                if (Consumer == null)
                {
                    FieldInfo.NormType = FieldInfo.DocValuesType_e.NUMERIC;
                    Consumer = new NumericDocValuesWriter(FieldInfo, DocState.DocWriter.bytesUsed, false);
                }
                Consumer.AddValue(DocState.DocID, Similarity.ComputeNorm(FieldState));
            }
        }

        internal void Flush(SegmentWriteState state, DocValuesConsumer normsWriter)
        {
            int docCount = state.SegmentInfo.DocCount;
            if (Consumer == null)
            {
                return; // null type - not omitted but not written -
                // meaning the only docs that had
                // norms hit exceptions (but indexed=true is set...)
            }
            Consumer.Finish(docCount);
            Consumer.Flush(state, normsWriter);
        }

        internal bool Empty
        {
            get
            {
                return Consumer == null;
            }
        }

        internal override void Abort()
        {
            //
        }
    }
}


text:
import pytest

import random

from pandas import Index
from geopandas import GeoDataFrame
from shapely.geometry import Point
from shapely.geometry.base import BaseGeometry
from shapely import wkt

from carto.exceptions import CartoException
from cartoframes.auth import Credentials
from cartoframes.io.managers.context_manager import ContextManager
from cartoframes.io.carto import read_carto, to_carto, copy_table, create_table_from_query

CREDENTIALS = Credentials('fake_user', 'fake_api_key')

def test_read_carto(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')
    cm_mock.return_value = GeoDataFrame({
        'cartodb_id': [1, 2, 3],
        'the_geom': [
            '010100000000000000000000000000000000000000',
            '010100000000000000000024400000000000002e40',
            '010100000000000000000034400000000000003e40'
        ]
    })
    expected = GeoDataFrame({
        'cartodb_id': [1, 2, 3],
        'the_geom': [
            Point([0, 0]),
            Point([10, 15]),
            Point([20, 30])
        ]
    }, geometry='the_geom')

    # When
    gdf = read_carto('__source__', CREDENTIALS)

    # Then
    cm_mock.assert_called_once_with('__source__', None, None, 3)
    assert expected.equals(gdf)
    assert gdf.crs == 'epsg:4326'

def test_read_carto_none_as_null_geom_value(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')
    cm_mock.return_value = GeoDataFrame({
        'cartodb_id': [1],
        'the_geom': [
            None
        ]
    })

    # When
    gdf = read_carto('__source__', CREDENTIALS)

    # Then
    expected = GeoDataFrame({
        'cartodb_id': [1],
        'the_geom': [
            None
        ]
    }, geometry='the_geom')

    cm_mock.assert_called_once_with('__source__', None, None, 3)
    assert expected.equals(gdf)
    assert gdf.crs == 'epsg:4326'

def test_read_carto_basegeometry_as_null_geom_value(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')
    cm_mock.return_value = GeoDataFrame({
        'cartodb_id': [1],
        'the_geom': [
            None
        ]
    })

    # When
    gdf = read_carto('__source__', CREDENTIALS, null_geom_value=BaseGeometry())

    # Then
    expected = GeoDataFrame({
        'cartodb_id': [1],
        'the_geom': [
            BaseGeometry()
        ]
    }, geometry='the_geom')

    cm_mock.assert_called_once_with('__source__', None, None, 3)
    assert expected.equals(gdf)
    assert gdf.crs == 'epsg:4326'

def test_read_carto_wrong_source(mocker):
    # When
    with pytest.raises(ValueError) as e:
        read_carto(1234)

    # Then
    assert str(e.value) == 'Wrong source. You should provide a valid table_name or SQL query.'

def test_read_carto_wrong_credentials(mocker):
    # When
    with pytest.raises(ValueError) as e:
        read_carto('__source__', 1234)

    # Then
    assert str(e.value) == 'Credentials attribute is required. Please pass a `Credentials` ' + \
                           'instance or use the `set_default_credentials` function.'

def test_read_carto_limit(mocker):
    # Given
    mocker.patch('cartoframes.utils.geom_utils.set_geometry')
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')

    # When
    read_carto('__source__', CREDENTIALS, limit=1)

    # Then
    cm_mock.assert_called_once_with('__source__', None, 1, 3)

def test_read_carto_retry_times(mocker):
    # Given
    mocker.patch('cartoframes.utils.geom_utils.set_geometry')
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')

    # When
    read_carto('__source__', CREDENTIALS, retry_times=1)

    # Then
    cm_mock.assert_called_once_with('__source__', None, None, 1)

def test_read_carto_schema(mocker):
    # Given
    mocker.patch('cartoframes.utils.geom_utils.set_geometry')
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')

    # When
    read_carto('__source__', CREDENTIALS, schema='__schema__')

    # Then
    cm_mock.assert_called_once_with('__source__', '__schema__', None, 3)

def test_read_carto_index_col_exists(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')
    cm_mock.return_value = GeoDataFrame({
        'cartodb_id': [1, 2, 3],
        'the_geom': [
            '010100000000000000000000000000000000000000',
            '010100000000000000000024400000000000002e40',
            '010100000000000000000034400000000000003e40'
        ]
    })
    expected = GeoDataFrame({
        'the_geom': [
            Point([0, 0]),
            Point([10, 15]),
            Point([20, 30])
        ]
    }, geometry='the_geom', index=Index([1, 2, 3], name='cartodb_id'))

    # When
    gdf = read_carto('__source__', CREDENTIALS, index_col='cartodb_id')

    # Then
    assert expected.equals(gdf)

def test_read_carto_index_col_not_exists(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')
    cm_mock.return_value = GeoDataFrame({
        'cartodb_id': [1, 2, 3],
        'the_geom': [
            '010100000000000000000000000000000000000000',
            '010100000000000000000024400000000000002e40',
            '010100000000000000000034400000000000003e40'
        ]
    })
    expected = GeoDataFrame({
        'cartodb_id': [1, 2, 3],
        'the_geom': [
            Point([0, 0]),
            Point([10, 15]),
            Point([20, 30])
        ]
    }, geometry='the_geom', index=Index([0, 1, 2], name='rename_index'))

    # When
    gdf = read_carto('__source__', CREDENTIALS, index_col='rename_index')

    # Then
    assert expected.equals(gdf)

def test_read_carto_decode_geom_false(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_to')
    cm_mock.return_value = GeoDataFrame({
        'cartodb_id': [1, 2, 3],
        'the_geom': [
            '010100000000000000000000000000000000000000',
            '010100000000000000000024400000000000002e40',
            '010100000000000000000034400000000000003e40'
        ]
    })
    expected = GeoDataFrame({
        'cartodb_id': [1, 2, 3],
        'the_geom': [
            '010100000000000000000000000000000000000000',
            '010100000000000000000024400000000000002e40',
            '010100000000000000000034400000000000003e40'
        ]
    })

    # When
    gdf = read_carto('__source__', CREDENTIALS, decode_geom=False)

    # Then
    assert expected.equals(gdf)

def test_to_carto(mocker):
    # Given
    table_name = '__table_name__'
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    cm_mock.return_value = table_name
    df = GeoDataFrame({'geometry': [Point([0, 0])]})

    # When
    norm_table_name = to_carto(df, table_name, CREDENTIALS, skip_quota_warning=True)

    # Then
    assert cm_mock.call_args[0][1] == table_name
    assert cm_mock.call_args[0][2] == 'fail'
    assert cm_mock.call_args[0][3] is True
    assert norm_table_name == table_name

def test_to_carto_non_4326(mocker):
    # Given
    table_name = '__table_name__'
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    cm_mock.return_value = table_name
    df = GeoDataFrame({'geometry': [Point([0, 0])]})
    df.crs = 'epsg:4269'

    # When
    norm_table_name = to_carto(df, table_name, CREDENTIALS, skip_quota_warning=True)

    # Then
    assert cm_mock.call_args[0][1] == table_name
    assert cm_mock.call_args[0][2] == 'fail'
    assert cm_mock.call_args[0][3] is True
    assert norm_table_name == table_name

def test_to_carto_quota_warning(mocker):
    class NoQuotaCredentials(Credentials):
        @property
        def me_data(self):
            return {
                'user_data': {
                    'remaining_byte_quota': 0
                }
            }

    # Given
    table_name = '__table_name__'
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    cm_mock.return_value = table_name
    df = GeoDataFrame({'geometry': [Point([0, 0])]})

    # Then
    with pytest.raises(CartoException):
        to_carto(df, table_name, NoQuotaCredentials('fake_user', 'fake_api_key'), skip_quota_warning=False)

def test_to_carto_quota_warning_skip(mocker):
    class NoQuotaCredentials(Credentials):
        @property
        def me_data(self):
            return {
                'user_data': {
                    'remaining_byte_quota': 0
                }
            }

    # Given
    table_name = '__table_name__'
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    cm_mock.return_value = table_name
    df = GeoDataFrame({'geometry': [Point([0, 0])]})

    # When
    norm_table_name = to_carto(df, table_name, NoQuotaCredentials('fake_user', 'fake_api_key'),
                               skip_quota_warning=True)

    # Then
    assert cm_mock.call_args[0][1] == table_name
    assert cm_mock.call_args[0][2] == 'fail'
    assert cm_mock.call_args[0][3] is True
    assert norm_table_name == table_name

def test_to_carto_two_geom_columns(mocker):
    # Given
    table_name = '__table_name__'
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    cm_mock.return_value = table_name
    df = GeoDataFrame({'geometry': [Point([0, 0])],
                       'the_geom': '010100000000000000000000000000000000000000'})

    # When
    norm_table_name = to_carto(df, table_name, CREDENTIALS, skip_quota_warning=True)

    # Then
    assert cm_mock.call_args[0][1] == table_name
    assert cm_mock.call_args[0][2] == 'fail'
    assert cm_mock.call_args[0][3] is True
    assert norm_table_name == table_name

def test_to_carto_two_geom_columns_and_geom_col(mocker):
    # Given
    table_name = '__table_name__'
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    cm_mock.return_value = table_name
    df = GeoDataFrame({'geometry': [Point([0, 0])],
                       'the_geom': '010100000000000000000000000000000000000000'})

    # When
    norm_table_name = to_carto(df, table_name, CREDENTIALS, geom_col='geometry', skip_quota_warning=True)

    # Then
    assert cm_mock.call_args[0][1] == table_name
    assert cm_mock.call_args[0][2] == 'fail'
    assert cm_mock.call_args[0][3] is True
    assert norm_table_name == table_name

def test_to_carto_chunks(mocker):
    # Given
    table_name = '__table_name__'
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    cm_mock.return_value = table_name

    size = 4000  # About 1MB (1150000 bytes)
    gdf = GeoDataFrame([
        ['Calle Gran Vía 46',
         round(random.uniform(10, 100), 2),
         round(random.uniform(100, 1000), 2),
         round(random.uniform(1000, 10000), 2),
         'POLYGON((-3.68831 40.42478, -3.68841 40.42478, -3.68841 40.42488, -3.68831 40.42478))']
        for _ in range(size)],
        columns=['address', 'value1', 'value2', 'value3', 'polygon']
    )
    gdf.set_geometry(gdf['polygon'].apply(wkt.loads), inplace=True)

    # When
    norm_table_name = to_carto(gdf, table_name, CREDENTIALS, max_upload_size=100000, skip_quota_warning=True)

    # Then
    assert cm_mock.call_count == 12  # 12 chunks as max_upload_size is 100000 bytes and we are uploading 1150000 bytes
    assert cm_mock.call_args[0][1] == table_name
    assert cm_mock.call_args[0][2] in ['fail', 'append']
    assert cm_mock.call_args[0][3] is True
    assert norm_table_name == table_name

def test_to_carto_wrong_dataframe(mocker):
    # When
    with pytest.raises(ValueError) as e:
        to_carto(1234, '', skip_quota_warning=True)

    # Then
    assert str(e.value) == 'Wrong dataframe. You should provide a valid DataFrame instance.'

def test_to_carto_wrong_table_name(mocker):
    # Given
    df = GeoDataFrame({'geometry': [Point([0, 0])]})

    # When
    with pytest.raises(ValueError) as e:
        to_carto(df, 1234, skip_quota_warning=True)

    # Then
    assert str(e.value) == 'Wrong table name. You should provide a valid table name.'

def test_to_carto_wrong_credentials(mocker):
    # Given
    df = GeoDataFrame({'geometry': [Point([0, 0])]})

    # When
    with pytest.raises(ValueError) as e:
        to_carto(df, '__table_name__', 1234, skip_quota_warning=True)

    # Then
    assert str(e.value) == 'Credentials attribute is required. Please pass a `Credentials` ' + \
                           'instance or use the `set_default_credentials` function.'

def test_to_carto_wrong_if_exists(mocker):
    # Given
    df = GeoDataFrame({'geometry': [Point([0, 0])]})

    # When
    with pytest.raises(ValueError) as e:
        to_carto(df, '__table_name__', if_exists='keep_calm', skip_quota_warning=True)

    # Then
    assert str(e.value) == 'Wrong option for the `if_exists` param. You should provide: fail, replace, append.'

def test_to_carto_if_exists_replace(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    df = GeoDataFrame({'geometry': [Point([0, 0])]})

    # When
    to_carto(df, '__table_name__', CREDENTIALS, if_exists='replace', skip_quota_warning=True)

    # Then
    assert cm_mock.call_args[0][2] == 'replace'

def test_to_carto_no_cartodbfy(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    df = GeoDataFrame({'geometry': [Point([0, 0])]})

    # When
    to_carto(df, '__table_name__', CREDENTIALS, cartodbfy=False, skip_quota_warning=True)

    # Then
    assert cm_mock.call_args[0][3] is False

def test_to_carto_replace_geometry(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'copy_from')
    df = GeoDataFrame({'geom': 'POINT(1 1)', 'geometry': [Point([0, 0])]})

    # When
    to_carto(df, '__table_name__', CREDENTIALS, geom_col='geom', cartodbfy=False, skip_quota_warning=True)

    # Then
    assert str(cm_mock.call_args[0][0]).strip() == 'the_geom\n0  POINT (1.00000 1.00000)'

def test_copy_table_wrong_table_name(mocker):
    # When
    with pytest.raises(ValueError) as e:
        copy_table(1234, '__new_table_name__')

    # Then
    assert str(e.value) == 'Wrong table name. You should provide a valid table name.'

def test_copy_table_wrong_new_table_name(mocker):
    # When
    with pytest.raises(ValueError) as e:
        copy_table('__table_name__', 1234)

    # Then
    assert str(e.value) == 'Wrong new table name. You should provide a valid table name.'

def test_copy_table_wrong_credentials(mocker):
    # When
    with pytest.raises(ValueError) as e:
        copy_table('__table_name__', '__new_table_name__', 1234)

    # Then
    assert str(e.value) == 'Credentials attribute is required. Please pass a `Credentials` ' + \
                           'instance or use the `set_default_credentials` function.'

def test_copy_table_wrong_if_exists(mocker):
    # When
    with pytest.raises(ValueError) as e:
        copy_table('__table_name__', '__new_table_name__', if_exists='keep_calm')

    # Then
    assert str(e.value) == 'Wrong option for the `if_exists` param. You should provide: fail, replace, append.'

def test_copy_table_no_cartodbfy(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'create_table_from_query')

    # When
    copy_table('__table_name__', '__new_table_name__', CREDENTIALS, cartodbfy=False)

    # Then
    assert cm_mock.call_args[0][3] is False

def test_copy_table_cartodbfy(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'create_table_from_query')

    # When
    copy_table('__table_name__', '__new_table_name__', CREDENTIALS, cartodbfy=True)

    # Then
    assert cm_mock.call_args[0][3] is True

def test_copy_table_cartodbfy_default(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'create_table_from_query')

    # When
    copy_table('__table_name__', '__new_table_name__', CREDENTIALS)

    # Then
    assert cm_mock.call_args[0][3] is True

def test_create_table_from_query_wrong_query(mocker):
    # When
    with pytest.raises(ValueError) as e:
        create_table_from_query('WRONG SQL QUERY', '__new_table_name__')

    # Then
    assert str(e.value) == 'Wrong query. You should provide a valid SQL query.'

def test_create_table_from_query_wrong_new_table_name(mocker):
    # When
    with pytest.raises(ValueError) as e:
        create_table_from_query('SELECT * FROM table', 1234)

    # Then
    assert str(e.value) == 'Wrong new table name. You should provide a valid table name.'

def test_create_table_from_query_wrong_credentials(mocker):
    # When
    with pytest.raises(ValueError) as e:
        create_table_from_query('SELECT * FROM table', '__new_table_name__', 1234)

    # Then
    assert str(e.value) == 'Credentials attribute is required. Please pass a `Credentials` ' + \
                           'instance or use the `set_default_credentials` function.'

def test_create_table_from_query_wrong_if_exists(mocker):
    # When
    with pytest.raises(ValueError) as e:
        create_table_from_query('SELECT * FROM table', '__new_table_name__', if_exists='keep_calm')

    # Then
    assert str(e.value) == 'Wrong option for the `if_exists` param. You should provide: fail, replace, append.'

def test_create_table_from_query_no_cartodbfy(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'create_table_from_query')

    # When
    create_table_from_query('SELECT * FROM table', '__new_table_name__', CREDENTIALS, cartodbfy=False)

    # Then
    assert cm_mock.call_args[0][3] is False

def test_create_table_from_query_cartodbfy(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'create_table_from_query')

    # When
    create_table_from_query('SELECT * FROM table', '__new_table_name__', CREDENTIALS, cartodbfy=True)

    # Then
    assert cm_mock.call_args[0][3] is True

def test_create_table_from_query_cartodbfy_default(mocker):
    # Given
    cm_mock = mocker.patch.object(ContextManager, 'create_table_from_query')

    # When
    create_table_from_query('SELECT * FROM table', '__new_table_name__', CREDENTIALS)

    # Then
    assert cm_mock.call_args[0][3] is True


text:
namespace Intacct.SDK.Functions.Common.Query.Comparison
{
    public abstract class AbstractString : AbstractComparison
    {
        public string Value;
    }
}


text:
<!DOCTYPE html>
<html>
  <head>
    <meta charset="utf-8">
    <title>A-Frame L-System Component - Hilbert Curve</title>
    <script src="../libs/aframe.js"></script>
    <script src="../libs/aframe-animation-component.js"></script>
    <script src="../libs/aframe-lsystem-component.js"></script>
  </head>
  <body>
    <a-scene>

      <a-assets>
        <a-mixin id="line" geometry="primitive: box; height: 0.2; width: 1; depth: 0.2;" material="metalness:0.2"></a-mixin>

        <a-mixin id="yellow" material="color: yellow;"></a-mixin>
        <a-mixin id="green" material="color: #528e1e;"></a-mixin>
        <a-mixin id="red" material="color: #8e311e;"></a-mixin>

      </a-assets>

      <a-entity id="hilbertcurve" lsystem="axiom: X; productions: X:^<XF^<XFX-F^>>XFX&F+>>XFX-F>X->; segmentMixins: F:green line; angle: 90.0; iterations: 2; translateAxis: X" rotation="0 45 0" position="0 0 -3">
      </a-entity>

</a-scene>
<script type="text/javascript">
  // doesnt work if i directly add the component above in the HTML :(
  document.getElementById('hilbertcurve').setAttribute('animation', "property: lsystem.angle; dur: 24000; loop: true; to: 450; easing: easeInOutCirc;");
</script>
</body>
</html>


text:
/*
 Erica Sadun, http://ericasadun.com
 iPhone Developer's Cookbook, 6.x Edition
 BSD License, Use at your own risk
 */

#import "DataManager.h"
#import "Utility.h"

@implementation DataManager
- (void) handleData
{
    NSMutableArray *newEntries = [NSMutableArray array];

    for (TreeNode *node in root.children)
        if (STREQ(node.key, @"entry"))
        {
            // Retrieve Name, Artist, Image
            NSString *name = [node leafForKey:@"im:name"];
            NSString *artist = [node leafForKey:@"im:artist"];
            NSString *imageAddress = [node leafForKey:@"im:image"];
            if (!name || !artist || !imageAddress) continue;

            NSData *data = [NSData dataWithContentsOfURL:[NSURL URLWithString:imageAddress]];
            UIImage *image = [UIImage imageWithData:data];
            if (!image) continue;

            // Store other info
            NSString *address = [node leafForKey:@"id"];
            NSString *price = [node leafForKey:@"im:price"];

            NSString *largeImageAddress = nil;
            UIImage  *largeImage = nil;
            NSArray  *images = [node leavesForKey:@"im:image"];
            if (images.count)
            {
                largeImageAddress = [images lastObject];
                data = [NSData dataWithContentsOfURL:[NSURL URLWithString:largeImageAddress]];
                largeImage = [UIImage imageWithData:data];
            }

            NSDictionary *entry = @{@"name":name, @"artist":artist, @"image":image, @"address":ENTRY(address), @"price":ENTRY(price), @"large image":ENTRY(largeImage)};
            [newEntries addObject:entry];
        }

    _entries = newEntries;

    if (_delegate && [_delegate respondsToSelector:@selector(dataIsReady:)])
        [_delegate performSelectorOnMainThread:@selector(dataIsReady:) withObject:self waitUntilDone:NO];
}

// Fetch data from iTunes
- (void) loadData
{
    NSString *rss = @"http://itunes.apple.com/us/rss/topalbums/limit=30/xml";
    NSOperationQueue *queue = [[NSOperationQueue alloc] init];
    [queue addOperationWithBlock:
     ^{
         root = [[XMLParser sharedInstance] parseXMLFromURL:[NSURL URLWithString:rss]];
         [[NSOperationQueue currentQueue] addOperationWithBlock:^{
             [self handleData];
         }];
     }];
}
@end


text:
package com.readystatesoftware.android;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.URLEncoder;

import org.apache.http.HttpResponse;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.DefaultHttpClient;
import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

import android.app.ListActivity;
import android.os.AsyncTask;
import android.os.Bundle;
import android.view.View;
import android.widget.AdapterView;
import android.widget.AdapterView.OnItemClickListener;
import android.widget.ListAdapter;
import android.widget.Toast;

/**
 * @author jgilfelt
 */
public class ExampleActivity extends ListActivity implements OnItemClickListener {

	@Override
	public void onCreate(Bundle savedInstanceState) {
		super.onCreate(savedInstanceState);
		new SearchTwitterTask().execute("#AndroidDev");
	}

	class SearchTwitterTask extends AsyncTask<String, Integer, String> {

		@Override
		protected String doInBackground(String... arg0) {
			String url = "http://search.twitter.com/search.json?q=" + URLEncoder.encode(arg0[0]);
			try {
				HttpClient client = new DefaultHttpClient();
				HttpGet request = new HttpGet(url);
				HttpResponse response = client.execute(request);
				BufferedReader rd = new BufferedReader(new InputStreamReader(response.getEntity().getContent()));
				String line = "";
				String json = "";
				while ((line = rd.readLine()) != null) {
					json += line;
				}
				rd.close();
				return json;
			} catch (Exception e) {
				e.printStackTrace();
				return null;
			}
		}

		@Override
		protected void onPostExecute(String result) {
			try {
				showData(result);
			} catch (JSONException e) {
				e.printStackTrace();
				Toast.makeText(ExampleActivity.this, "something went wrong", Toast.LENGTH_SHORT).show();
			}
		}

	}

	private void showData(String json) throws JSONException {
		JSONObject o = new JSONObject(json);
		JSONArray data = o.getJSONArray("results");

		ListAdapter adapter = new JSONArrayAdapter(this,
				data,													// JSONArray data
				android.R.layout.simple_list_item_2, 					// a layout resource to display a row
				new String[] {"from_user", "text"},						// field names from JSONObjects
				new int[] {android.R.id.text1, android.R.id.text2},		// corresponding View ids to map field names to
				"id"													// id field from JSONObjects (optional)
		);

		getListView().setAdapter(adapter);
		getListView().setOnItemClickListener(this);
	}

	@Override
	public void onItemClick(AdapterView<?> parent, View view, int position, long id) {
		Toast.makeText(ExampleActivity.this, "clicked tweet id " + id, Toast.LENGTH_SHORT).show();
	}

}


text:
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
	<modelVersion>4.0.0</modelVersion>

	<parent>
		<groupId>org.teknux</groupId>
		<artifactId>jetty-bootstrap-parent</artifactId>
		<version>1.0.15-SNAPSHOT</version>
	</parent>

	<artifactId>jetty-bootstrap-logs</artifactId>
	<packaging>jar</packaging>

	<name>Jetty Bootstrap : Logs Dependencies</name>
	<url>http://jetty-bootstrap.io</url>

	<dependencies>
		<dependency>
			<groupId>org.slf4j</groupId>
			<artifactId>jul-to-slf4j</artifactId>
			<version>${slf4j.version}</version>
		</dependency>
		<dependency>
			<groupId>ch.qos.logback</groupId>
			<artifactId>logback-classic</artifactId>
			<version>${logback.version}</version>
		</dependency>
		<!-- For condition in logback.xml -->
		<dependency>
			<groupId>org.codehaus.janino</groupId>
			<artifactId>janino</artifactId>
			<version>${janino.version}</version>
		</dependency>
	</dependencies>

	<properties>
		<logback.version>1.0.13</logback.version>
		<janino.version>2.6.1</janino.version>
	</properties>

</project>


text:
@implementation ConnectableDevice
{
    NSMutableDictionary *_services;
}

@synthesize serviceDescription = _consolidatedServiceDescription;
@synthesize delegate = _delegate;
@synthesize id = _id;

- (instancetype) init
{
    self = [super init];

    if (self)
    {
        _consolidatedServiceDescription = [ServiceDescription new];
        _services = [[NSMutableDictionary alloc] init];
    }

    return self;
}

- (instancetype) initWithDescription:(ServiceDescription *)description
{
    self = [self init];

    if (self)
    {
        _consolidatedServiceDescription = description;
    }

    return self;
}

- (instancetype) initWithJSONObject:(NSDictionary *)dict
{
    self = [self init];

    if (self)
    {
        _id = dict[@"id"];
        _lastKnownIPAddress = dict[@"lastKnownIPAddress"];
        _lastSeenOnWifi = dict[@"lastSeenOnWifi"];

        id lastConnected = dict[@"lastConnected"];
        if (lastConnected && ![lastConnected isKindOfClass:[NSNull class]])
            _lastConnected = [lastConnected doubleValue];

        id lastDetection = dict[@"lastDetection"];
        if (lastDetection && ![lastDetection isKindOfClass:[NSNull class]])
            _lastDetection = [lastDetection doubleValue];

        if (!self.address)
            _consolidatedServiceDescription.address = _lastKnownIPAddress;
    }

    return self;
}

- (NSDictionary *) toJSONObject
{
    NSMutableDictionary *jsonObject = [NSMutableDictionary new];

    if (self.id) jsonObject[@"id"] = self.id;
    if (self.friendlyName) jsonObject[@"friendlyName"] = self.friendlyName;
    if (self.lastKnownIPAddress) jsonObject[@"lastKnownIPAddress"] = self.lastKnownIPAddress;
    if (self.lastSeenOnWifi) jsonObject[@"lastSeenOnWifi"] = self.lastSeenOnWifi;
    if (self.lastConnected) jsonObject[@"lastConnected"] = @(self.lastConnected);
    if (self.lastDetection) jsonObject[@"lastDetection"] = @(self.lastDetection);

    NSMutableDictionary *services = [NSMutableDictionary new];

    [self.services enumerateObjectsUsingBlock:^(DeviceService *service, NSUInteger idx, BOOL *stop)
    {
        NSDictionary *serviceJSON = [service toJSONObject];
        [services setObject:serviceJSON forKey:service.serviceDescription.UUID];
    }];

    if (services.count > 0)
        jsonObject[@"services"] = [NSDictionary dictionaryWithDictionary:services];

    return jsonObject;
}

+ (instancetype) connectableDeviceWithDescription:(ServiceDescription *)description
{
    return [[ConnectableDevice alloc] initWithDescription:description];
}

- (void) setDelegate:(id<ConnectableDeviceDelegate>)delegate
{
    _delegate = delegate;

    if (_delegate && [self.delegate respondsToSelector:@selector(connectableDeviceConnectionRequired:forService:)])
    {
        [_services enumerateKeysAndObjectsUsingBlock:^(id key, DeviceService *service, BOOL *stop) {
            if (service.isConnectable && !service.connected)
                [_delegate connectableDeviceConnectionRequired:self forService:service];
        }];
    }
}

#pragma mark - General info

- (NSString *) id
{
    if (!_id)
        _id = [[CTGuid randomGuid] stringValueWithFormat:CTGuidFormatDashed];

    return _id;
}

- (NSString *) address
{
    return _consolidatedServiceDescription.address;
}

- (NSString *) friendlyName
{
    return _consolidatedServiceDescription.friendlyName;
}

- (NSString *) modelName
{
    return _consolidatedServiceDescription.modelName;
}

- (NSString *) modelNumber
{
    return _consolidatedServiceDescription.modelNumber;
}

- (NSString *) connectedServiceNames
{
    __block NSString *serviceNames = @"";

    if (_services.count == 0)
        return serviceNames;

    [_services enumerateKeysAndObjectsUsingBlock:^(NSString * key, DeviceService *service, BOOL *stop) {
        serviceNames = [serviceNames stringByAppendingString:[NSString stringWithFormat:@"%@, ", key]];
    }];

    serviceNames = [serviceNames substringToIndex:serviceNames.length - 2];

    return serviceNames;
}

- (BOOL) connected
{
    __block int connectedCount = 0;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, DeviceService *service, BOOL *stop)
    {
        if (!service.isConnectable)
            connectedCount++;
        else
        {
            if (service.connected)
                connectedCount++;
        }
    }];

    return connectedCount >= _services.count;
}

- (void) connect
{
    if (self.connected)
    {
        dispatch_on_main(^{ [self.delegate connectableDeviceReady:self]; });
    } else
    {
        [_services enumerateKeysAndObjectsUsingBlock:^(id key, DeviceService *service, BOOL *stop)
        {
            if (!service.connected)
                [service connect];
        }];
    }

    [[NSNotificationCenter defaultCenter] addObserver:self selector:@selector(disconnect) name:kConnectSDKWirelessSSIDChanged object:nil];
}

- (void) disconnect
{
    [_services enumerateKeysAndObjectsUsingBlock:^(id key, DeviceService *service, BOOL *stop)
    {
        if (service.connected)
            [service disconnect];
    }];

    [[NSNotificationCenter defaultCenter] removeObserver:self name:kConnectSDKWirelessSSIDChanged object:nil];

    dispatch_on_main(^{ [self.delegate connectableDeviceDisconnected:self withError:nil]; });
}

- (BOOL) isConnectable
{
    __block BOOL connectable = NO;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, DeviceService *service, BOOL *stop)
    {
        if (service.isConnectable)
        {
            connectable = YES;
            *stop = YES;
        }
    }];

    return connectable;
}

- (int) connectedServiceCount
{
    __block int count = 0;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, DeviceService *service, BOOL *stop)
    {
        if ([service isConnectable])
        {
            if (service.connected)
                count++;
        } else
        {
            count++;
        }

    }];

    return count;
}

- (int) connectableServiceCount
{
    __block int count = 0;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, DeviceService *service, BOOL *stop)
    {
        if ([service isConnectable])
            count++;
    }];

    return count;
}

#pragma mark - Service management

- (NSArray *) services
{
    return [_services allValues];
}

- (BOOL) hasServices
{
    return _services.count > 0;
}

- (void) addService:(DeviceService *)service
{
    DeviceService *existingService = [_services objectForKey:service.serviceName];

    NSArray *oldCapabilities = self.capabilities;

    if (existingService)
    {
        if (service.serviceDescription.lastDetection > existingService.serviceDescription.lastDetection)
        {
            if (existingService.connected)
                [existingService disconnect];

            DLog(@"Removing %@ (%@)", existingService.serviceDescription.friendlyName, existingService.serviceName);
            [self removeServiceWithId:existingService.serviceName];
        } else
        {
            DLog(@"Ignoring %@ (%@)", service.serviceDescription.friendlyName, service.serviceName);
            return;
        }
    }

    [_services setObject:service forKey:service.serviceName];

    if (service.delegate == nil)
        service.delegate = self;

    if (service.isConnectable && !service.connected)
    {
        if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDeviceConnectionRequired:forService:)])
            dispatch_on_main(^{ [_delegate connectableDeviceConnectionRequired:self forService:service]; });
    }

    [self updateCapabilitiesList:oldCapabilities];

    [self updateConsolidatedServiceDescription:service.serviceDescription];
}

- (void) removeServiceWithId:(NSString *)serviceId
{
    DeviceService *service = [_services objectForKey:serviceId];

    if (service == nil)
        return;

    NSArray *oldCapabilities = self.capabilities;

    [service disconnect];

    [_services removeObjectForKey:serviceId];

    [self updateCapabilitiesList:oldCapabilities];
}

- (void) updateCapabilitiesList:(NSArray *)oldCapabilities
{
    NSArray *newCapabilities = self.capabilities;

    NSMutableArray *removedCapabilities = [NSMutableArray new];

    [oldCapabilities enumerateObjectsUsingBlock:^(NSString *capability, NSUInteger idx, BOOL *stop) {
        if (![newCapabilities containsObject:capability])
            [removedCapabilities addObject:capability];
    }];

    NSMutableArray *addedCapabilities = [NSMutableArray new];

    [newCapabilities enumerateObjectsUsingBlock:^(NSString *capability, NSUInteger idx, BOOL *stop) {
        if (![oldCapabilities containsObject:capability])
            [addedCapabilities addObject:capability];
    }];

    NSArray *added = [NSArray arrayWithArray:addedCapabilities];
    NSArray *removed = [NSArray arrayWithArray:removedCapabilities];

    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDevice:capabilitiesAdded:removed:)])
        dispatch_on_main(^{ [self.delegate connectableDevice:self capabilitiesAdded:added removed:removed]; });
}

- (void) updateConsolidatedServiceDescription:(ServiceDescription *)serviceDescription
{
    if (serviceDescription.address)
        _consolidatedServiceDescription.address = serviceDescription.address;

    if (serviceDescription.friendlyName)
        _consolidatedServiceDescription.friendlyName = serviceDescription.friendlyName;

    if (serviceDescription.modelName)
        _consolidatedServiceDescription.modelName = serviceDescription.modelName;

    if (serviceDescription.modelNumber)
        _consolidatedServiceDescription.modelNumber = serviceDescription.modelNumber;
}

- (DeviceService *)serviceWithName:(NSString *)serviceId
{
    __block DeviceService *foundService;

    [_services enumerateKeysAndObjectsUsingBlock:^(NSString *id, DeviceService *service, BOOL *stop)
    {
        if ([id isEqualToString:serviceId])
        {
            foundService = service;
            *stop = YES;
        }
    }];

    return foundService;
}

#pragma mark - DeviceServiceDelegate
#pragma mark Connection

- (void)deviceServiceConnectionRequired:(DeviceService *)service
{
    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDeviceConnectionRequired:forService:)])
        dispatch_on_main(^{ [self.delegate connectableDeviceConnectionRequired:self forService:service]; });
}

- (void)deviceServiceConnectionSuccess:(DeviceService *)service
{
    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDeviceConnectionSuccess:forService:)])
        dispatch_on_main(^{ [self.delegate connectableDeviceConnectionSuccess:self forService:service]; });

    if (self.connected)
    {
        [[[DiscoveryManager sharedManager] deviceStore] addDevice:self];

        dispatch_on_main(^{ [self.delegate connectableDeviceReady:self]; });

        self.lastConnected = [[NSDate date] timeIntervalSince1970];
    }
}

- (void)deviceService:(DeviceService *)service didFailConnectWithError:(NSError *)error
{
    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDevice:connectionFailedWithError:)])
        dispatch_on_main(^{ [self.delegate connectableDevice:self connectionFailedWithError:error]; });

    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDevice:service:didFailConnectWithError:)])
        dispatch_on_main(^{ [self.delegate connectableDevice:self service:service didFailConnectWithError:error]; });
}

- (void)deviceService:(DeviceService *)service disconnectedWithError:(NSError *)error
{
    // TODO: need to aggregate errors between disconnects
    if ([self connectedServiceCount] == 0 || _services.count == 0)
        dispatch_on_main(^{ [self.delegate connectableDeviceDisconnected:self withError:error]; });

    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDevice:service:disconnectedWithError:)])
        dispatch_on_main(^{ [self.delegate connectableDevice:self service:service disconnectedWithError:error]; });
}

#pragma mark Pairing

- (void)deviceService:(DeviceService *)service pairingRequiredOfType:(DeviceServicePairingType)pairingType withData:(id)pairingData
{
    if (self.delegate)
    {
        if ([self.delegate respondsToSelector:@selector(connectableDevice:service:pairingRequiredOfType:withData:)])
            dispatch_on_main(^{ [self.delegate connectableDevice:self service:service pairingRequiredOfType:pairingType withData:pairingData]; });
        else
        {
            if (pairingType == DeviceServicePairingTypeAirPlayMirroring)
                [(UIAlertView *)pairingData show];
        }
    }
}

- (void)deviceServicePairingSuccess:(DeviceService *)service
{
    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDevicePairingSuccess:service:)])
        dispatch_on_main(^{ [self.delegate connectableDevicePairingSuccess:self service:service]; });
}

- (void)deviceService:(DeviceService *)service pairingFailedWithError:(NSError *)error
{
    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDevice:service:pairingFailedWithError:)])
        dispatch_on_main(^{ [self.delegate connectableDevice:self service:service pairingFailedWithError:error]; });
}

#pragma mark Capability updates

- (void) deviceService:(DeviceService *)service capabilitiesAdded:(NSArray *)added removed:(NSArray *)removed
{
    [[DiscoveryManager sharedManager] connectableDevice:self capabilitiesAdded:added removed:removed];

    if (self.delegate && [self.delegate respondsToSelector:@selector(connectableDevice:capabilitiesAdded:removed:)])
        dispatch_on_main(^{ [self.delegate connectableDevice:self capabilitiesAdded:added removed:removed]; });
}

#pragma mark - Capabilities

- (NSArray *) capabilities
{
    NSMutableArray *caps = [NSMutableArray new];

    [self.services enumerateObjectsUsingBlock:^(DeviceService *service, NSUInteger serviceIdx, BOOL *serviceStop)
    {
        [service.capabilities enumerateObjectsUsingBlock:^(NSString *capability, NSUInteger capabilityIdx, BOOL *capabilityStop)
        {
            if (![caps containsObject:capability])
                [caps addObject:capability];
        }];
    }];

    return [NSArray arrayWithArray:caps];
}

- (BOOL) hasCapability:(NSString *)capability
{
    __block BOOL hasCap = NO;

    [self.services enumerateObjectsUsingBlock:^(DeviceService *service, NSUInteger idx, BOOL *stop)
    {
        if ([service hasCapability:capability])
        {
            hasCap = YES;
            *stop = YES;
        }
    }];

    return hasCap;
}

- (BOOL) hasCapabilities:(NSArray *)capabilities
{
    __block BOOL hasCaps = YES;

    [capabilities enumerateObjectsUsingBlock:^(NSString *capability, NSUInteger idx, BOOL *stop)
    {
        if (![self hasCapability:capability])
        {
            hasCaps = NO;
            *stop = YES;
        }
    }];

    return hasCaps;
}

- (BOOL) hasAnyCapability:(NSArray *)capabilities
{
    __block BOOL hasCap = NO;

    [self.services enumerateObjectsUsingBlock:^(DeviceService *service, NSUInteger idx, BOOL *stop)
    {
        if ([service hasAnyCapability:capabilities])
        {
            hasCap = YES;
            *stop = YES;
        }
    }];

    return hasCap;
}

- (id<Launcher>) launcher
{
    __block id<Launcher> foundLauncher;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(launcher)])
            return;

        id<Launcher> launcher = [service launcher];

        if (launcher)
        {
            if (foundLauncher)
            {
                if (launcher.launcherPriority > foundLauncher.launcherPriority)
                {
                    foundLauncher = launcher;
                }
            } else
            {
                foundLauncher = launcher;
            }
        }
    }];

    return foundLauncher;
}

- (id<ExternalInputControl>)externalInputControl
{
    __block id<ExternalInputControl> foundExternalInputControl;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(externalInputControl)])
            return;

        id<ExternalInputControl> externalInputControl = [service externalInputControl];

        if (externalInputControl)
        {
            if (foundExternalInputControl)
            {
                if (externalInputControl.externalInputControlPriority > foundExternalInputControl.externalInputControlPriority)
                {
                    foundExternalInputControl = externalInputControl;
                }
            } else
            {
                foundExternalInputControl = externalInputControl;
            }
        }
    }];

    return foundExternalInputControl;
}

- (id<MediaPlayer>) mediaPlayer
{
    __block id<MediaPlayer> foundPlayer;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(mediaPlayer)])
            return;

        id<MediaPlayer> player = [service mediaPlayer];

        if (player)
        {
            if (foundPlayer)
            {
                if (player.mediaPlayerPriority > foundPlayer.mediaPlayerPriority)
                {
                    foundPlayer = player;
                }
            } else
            {
                foundPlayer = player;
            }
        }
    }];

    return foundPlayer;
}

- (id<MediaControl>) mediaControl
{
    __block id<MediaControl> foundMediaControl;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(mediaControl)])
            return;

        id<MediaControl> mediaControl = [service mediaControl];

        if (mediaControl)
        {
            if (foundMediaControl)
            {
                if (mediaControl.mediaControlPriority > foundMediaControl.mediaControlPriority)
                {
                    foundMediaControl = mediaControl;
                }
            } else
            {
                foundMediaControl = mediaControl;
            }
        }
    }];

    return foundMediaControl;
}

- (id<VolumeControl>)volumeControl
{
    __block id<VolumeControl> foundVolume;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(volumeControl)])
            return;

        id<VolumeControl> volume = [service volumeControl];

        if (volume)
        {
            if (foundVolume)
            {
                if (volume.volumeControlPriority > foundVolume.volumeControlPriority)
                {
                    foundVolume = volume;
                }
            } else
            {
                foundVolume = volume;
            }
        }
    }];

    return foundVolume;
}

- (id<TVControl>)tvControl
{
    __block id<TVControl> foundTV;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(tvControl)])
            return;

        id<TVControl> tv = [service tvControl];

        if (tv)
        {
            if (foundTV)
            {
                if (tv.tvControlPriority > foundTV.tvControlPriority)
                {
                    foundTV = tv;
                }
            } else
            {
                foundTV = tv;
            }
        }
    }];

    return foundTV;
}

- (id<KeyControl>) keyControl
{
    __block id<KeyControl> foundKeyControl;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(keyControl)])
            return;

        id<KeyControl> keyControl = [service keyControl];

        if (keyControl)
        {
            if (foundKeyControl)
            {
                if (keyControl.keyControlPriority > foundKeyControl.keyControlPriority)
                {
                    foundKeyControl = keyControl;
                }
            } else
            {
                foundKeyControl = keyControl;
            }
        }
    }];

    return foundKeyControl;
}

- (id<TextInputControl>) textInputControl
{
    __block id<TextInputControl> foundTextInput;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(textInputControl)])
            return;

        id<TextInputControl> textInput = [service textInputControl];

        if (textInput)
        {
            if (foundTextInput)
            {
                if (textInput.textInputControlPriority > foundTextInput.textInputControlPriority)
                {
                    foundTextInput = textInput;
                }
            } else
            {
                foundTextInput = textInput;
            }
        }
    }];

    return foundTextInput;
}

- (id<MouseControl>)mouseControl
{
    __block id<MouseControl> foundMouse;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(mouseControl)])
            return;

        id<MouseControl> mouse = [service mouseControl];

        if (mouse)
        {
            if (foundMouse)
            {
                if (mouse.mouseControlPriority > foundMouse.mouseControlPriority)
                {
                    foundMouse = mouse;
                }
            } else
            {
                foundMouse = mouse;
            }
        }
    }];

    return foundMouse;
}

- (id<PowerControl>)powerControl
{
    __block id<PowerControl> foundPower;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(powerControl)])
            return;

        id<PowerControl> power = [service powerControl];

        if (power)
        {
            if (foundPower)
            {
                if (power.powerControlPriority > foundPower.powerControlPriority)
                {
                    foundPower = power;
                }
            } else
            {
                foundPower = power;
            }
        }
    }];

    return foundPower;
}

- (id<ToastControl>) toastControl
{
    __block id<ToastControl> foundToastControl;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(toastControl)])
            return;

        id<ToastControl> toastControl = [service toastControl];

        if (toastControl)
        {
            if (foundToastControl)
            {
                if (toastControl.toastControlPriority > foundToastControl.toastControlPriority)
                {
                    foundToastControl = toastControl;
                }
            } else
            {
                foundToastControl = toastControl;
            }
        }
    }];

    return foundToastControl;
}

- (id<WebAppLauncher>) webAppLauncher
{
    __block id<WebAppLauncher> foundWebAppLauncher;

    [_services enumerateKeysAndObjectsUsingBlock:^(id key, id service, BOOL *stop)
    {
        if (![service respondsToSelector:@selector(webAppLauncher)])
            return;

        id<WebAppLauncher> webAppLauncher = [service webAppLauncher];

        if (webAppLauncher)
        {
            if (foundWebAppLauncher)
            {
                if (webAppLauncher.webAppLauncherPriority > foundWebAppLauncher.webAppLauncherPriority)
                {
                    foundWebAppLauncher = webAppLauncher;
                }
            } else
            {
                foundWebAppLauncher = webAppLauncher;
            }
        }
    }];

    return foundWebAppLauncher;
}

@end


text:
#include "router_core_private.h"
#include <qpid/dispatch/amqp.h>
#include <stdio.h>

static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_link_flow_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_link_check_credit_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_send_to_CT(qdr_core_t *core, qdr_action_t *action, bool discard);
static void qdr_update_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard);

//==================================================================================
// Internal Functions
//==================================================================================

//==================================================================================
// Interface Functions
//==================================================================================

qdr_delivery_t *qdr_link_deliver(qdr_link_t *link, qd_message_t *msg, qd_iterator_t *ingress,
                                 bool settled, qd_bitmask_t *link_exclusion)
{
    qdr_action_t   *action = qdr_action(qdr_link_deliver_CT, "link_deliver");
    qdr_delivery_t *dlv    = new_qdr_delivery_t();

    ZERO(dlv);
    sys_atomic_init(&dlv->ref_count, 1); // referenced by the action
    dlv->link           = link;
    dlv->msg            = msg;
    dlv->to_addr        = 0;
    dlv->origin         = ingress;
    dlv->settled        = settled;
    dlv->presettled     = settled;
    dlv->link_exclusion = link_exclusion;

    action->args.connection.delivery = dlv;
    qdr_action_enqueue(link->core, action);
    return dlv;
}

qdr_delivery_t *qdr_link_deliver_to(qdr_link_t *link, qd_message_t *msg,
                                    qd_iterator_t *ingress, qd_iterator_t *addr,
                                    bool settled, qd_bitmask_t *link_exclusion)
{
    qdr_action_t   *action = qdr_action(qdr_link_deliver_CT, "link_deliver");
    qdr_delivery_t *dlv    = new_qdr_delivery_t();

    ZERO(dlv);
    sys_atomic_init(&dlv->ref_count, 1); // referenced by the action
    dlv->link           = link;
    dlv->msg            = msg;
    dlv->to_addr        = addr;
    dlv->origin         = ingress;
    dlv->settled        = settled;
    dlv->presettled     = settled;
    dlv->link_exclusion = link_exclusion;

    action->args.connection.delivery = dlv;
    qdr_action_enqueue(link->core, action);
    return dlv;
}

qdr_delivery_t *qdr_link_deliver_to_routed_link(qdr_link_t *link, qd_message_t *msg, bool settled,
                                                const uint8_t *tag, int tag_length)
{
    if (tag_length > 32)
        return 0;

    qdr_action_t   *action = qdr_action(qdr_link_deliver_CT, "link_deliver");
    qdr_delivery_t *dlv    = new_qdr_delivery_t();

    ZERO(dlv);
    sys_atomic_init(&dlv->ref_count, 1); // referenced by the action
    dlv->link       = link;
    dlv->msg        = msg;
    dlv->settled    = settled;
    dlv->presettled = settled;

    action->args.connection.delivery = dlv;
    action->args.connection.tag_length = tag_length;
    memcpy(action->args.connection.tag, tag, tag_length);
    qdr_action_enqueue(link->core, action);
    return dlv;
}

void qdr_link_process_deliveries(qdr_core_t *core, qdr_link_t *link, int credit)
{
    qdr_connection_t *conn = link->conn;
    qdr_delivery_t   *dlv;
    bool              drained = false;
    int               offer   = -1;
    bool              settled = false;

    if (link->link_direction == QD_OUTGOING) {
        while (credit > 0 && !drained) {
            sys_mutex_lock(conn->work_lock);
            dlv = DEQ_HEAD(link->undelivered);
            if (dlv) {
                DEQ_REMOVE_HEAD(link->undelivered);
                settled = dlv->settled;
                if (!settled) {
                    DEQ_INSERT_TAIL(link->unsettled, dlv);
                    dlv->where = QDR_DELIVERY_IN_UNSETTLED;
                } else
                    dlv->where = QDR_DELIVERY_NOWHERE;

                credit--;
                link->total_deliveries++;
                offer = DEQ_SIZE(link->undelivered);
            } else
                drained = true;
            sys_mutex_unlock(conn->work_lock);

            if (dlv) {
                link->credit_to_core--;
                core->deliver_handler(core->user_context, link, dlv, settled);
                if (settled)
                    qdr_delivery_decref(dlv);
            }
        }

        if (drained)
            core->drained_handler(core->user_context, link);
        else if (offer != -1)
            core->offer_handler(core->user_context, link, offer);
    }

    //
    // Handle disposition/settlement updates
    //
    qdr_delivery_ref_list_t updated_deliveries;
    sys_mutex_lock(conn->work_lock);
    DEQ_MOVE(link->updated_deliveries, updated_deliveries);
    sys_mutex_unlock(conn->work_lock);

    qdr_delivery_ref_t *ref = DEQ_HEAD(updated_deliveries);
    while (ref) {
        core->delivery_update_handler(core->user_context, ref->dlv, ref->dlv->disposition, ref->dlv->settled);
        qdr_delivery_decref(ref->dlv);
        qdr_del_delivery_ref(&updated_deliveries, ref);
        ref = DEQ_HEAD(updated_deliveries);
    }
}

void qdr_link_flow(qdr_core_t *core, qdr_link_t *link, int credit, bool drain_mode)
{
    qdr_action_t *action = qdr_action(qdr_link_flow_CT, "link_flow");

    //
    // Compute the number of credits now available that we haven't yet given
    // incrementally to the router core.  i.e. convert absolute credit to
    // incremental credit.
    //
    credit -= link->credit_to_core;
    if (credit < 0)
        credit = 0;
    link->credit_to_core += credit;

    action->args.connection.link   = link;
    action->args.connection.credit = credit;
    action->args.connection.drain  = drain_mode;

    qdr_action_enqueue(core, action);
}

void qdr_link_check_credit(qdr_core_t *core, qdr_link_t *link)
{
    qdr_action_t *action = qdr_action(qdr_link_check_credit_CT, "link_check_credit");
    action->args.connection.link = link;
    qdr_action_enqueue(core, action);
}

void qdr_send_to1(qdr_core_t *core, qd_message_t *msg, qd_iterator_t *addr, bool exclude_inprocess, bool control)
{
    qdr_action_t *action = qdr_action(qdr_send_to_CT, "send_to");
    action->args.io.address           = qdr_field_from_iter(addr);
    action->args.io.message           = qd_message_copy(msg);
    action->args.io.exclude_inprocess = exclude_inprocess;
    action->args.io.control           = control;

    qdr_action_enqueue(core, action);
}

void qdr_send_to2(qdr_core_t *core, qd_message_t *msg, const char *addr, bool exclude_inprocess, bool control)
{
    qdr_action_t *action = qdr_action(qdr_send_to_CT, "send_to");
    action->args.io.address           = qdr_field(addr);
    action->args.io.message           = qd_message_copy(msg);
    action->args.io.exclude_inprocess = exclude_inprocess;
    action->args.io.control           = control;

    qdr_action_enqueue(core, action);
}

void qdr_delivery_update_disposition(qdr_core_t *core, qdr_delivery_t *delivery, uint64_t disposition,
                                     bool settled, bool ref_given)
{
    qdr_action_t *action = qdr_action(qdr_update_delivery_CT, "update_delivery");
    action->args.delivery.delivery    = delivery;
    action->args.delivery.disposition = disposition;
    action->args.delivery.settled     = settled;

    //
    // The delivery's ref_count must be incremented to protect its travels into the
    // core thread.  If the caller has given its reference to us, we can simply use
    // the given ref rather than increment a new one.
    //
    if (!ref_given)
        qdr_delivery_incref(delivery);

    qdr_action_enqueue(core, action);
}

void qdr_delivery_set_context(qdr_delivery_t *delivery, void *context)
{
    delivery->context = context;
}

void *qdr_delivery_get_context(qdr_delivery_t *delivery)
{
    return delivery->context;
}

void qdr_delivery_incref(qdr_delivery_t *delivery)
{
    sys_atomic_inc(&delivery->ref_count);
}

void qdr_delivery_decref(qdr_delivery_t *delivery)
{
    qdr_link_t *link      = delivery->link;
    uint32_t    ref_count = sys_atomic_dec(&delivery->ref_count);
    assert(ref_count > 0);

    if (ref_count == 1) {
        if (delivery->msg)
            qd_message_free(delivery->msg);

        if (delivery->to_addr)
            qd_iterator_free(delivery->to_addr);

        if (delivery->tracking_addr) {
            delivery->tracking_addr->outstanding_deliveries[delivery->tracking_addr_bit]--;
            delivery->tracking_addr->tracked_deliveries--;
            delivery->tracking_addr = 0;
        }

        if (link) {
            if (delivery->presettled)
                link->presettled_deliveries++;
            else if (delivery->disposition == PN_ACCEPTED)
                link->accepted_deliveries++;
            else if (delivery->disposition == PN_REJECTED)
                link->rejected_deliveries++;
            else if (delivery->disposition == PN_RELEASED)
                link->released_deliveries++;
            else if (delivery->disposition == PN_MODIFIED)
                link->modified_deliveries++;
        }

        qd_bitmask_free(delivery->link_exclusion);
        free_qdr_delivery_t(delivery);
    }
}

void qdr_delivery_tag(const qdr_delivery_t *delivery, const char **tag, int *length)
{
    *tag    = (const char*) delivery->tag;
    *length = delivery->tag_length;
}

qd_message_t *qdr_delivery_message(const qdr_delivery_t *delivery)
{
    return delivery->msg;
}

//==================================================================================
// In-Thread Functions
//==================================================================================

void qdr_delivery_release_CT(qdr_core_t *core, qdr_delivery_t *dlv)
{
    bool push = dlv->disposition != PN_RELEASED;

    dlv->disposition = PN_RELEASED;
    dlv->settled = true;
    bool moved = qdr_delivery_settled_CT(core, dlv);

    if (push || moved)
        qdr_delivery_push_CT(core, dlv);

    //
    // Remove the unsettled reference
    //
    if (moved)
        qdr_delivery_decref(dlv);
}

void qdr_delivery_failed_CT(qdr_core_t *core, qdr_delivery_t *dlv)
{
    bool push = dlv->disposition != PN_MODIFIED;

    dlv->disposition = PN_MODIFIED;
    dlv->settled = true;
    bool moved = qdr_delivery_settled_CT(core, dlv);

    if (push || moved)
        qdr_delivery_push_CT(core, dlv);

    //
    // Remove the unsettled reference
    //
    if (moved)
        qdr_delivery_decref(dlv);
}

bool qdr_delivery_settled_CT(qdr_core_t *core, qdr_delivery_t *dlv)
{
    //
    // Remove a delivery from its unsettled list.  Side effects include issuing
    // replacement credit and visiting the link-quiescence algorithm
    //
    qdr_link_t       *link  = dlv->link;
    qdr_connection_t *conn  = link ? link->conn : 0;
    bool              moved = false;

    if (!link || !conn)
        return false;

    //
    // The lock needs to be acquired only for outgoing links
    //
    if (link->link_direction == QD_OUTGOING)
        sys_mutex_lock(conn->work_lock);

    if (dlv->where == QDR_DELIVERY_IN_UNSETTLED) {
        DEQ_REMOVE(link->unsettled, dlv);
        dlv->where = QDR_DELIVERY_NOWHERE;
        moved = true;
    }

    if (link->link_direction == QD_OUTGOING)
        sys_mutex_unlock(conn->work_lock);

    if (dlv->tracking_addr) {
        dlv->tracking_addr->outstanding_deliveries[dlv->tracking_addr_bit]--;
        dlv->tracking_addr->tracked_deliveries--;
        dlv->tracking_addr = 0;
    }

    //
    // If this is an incoming link and it is not link-routed or inter-router, issue
    // one replacement credit on the link.  Note that credit on inter-router links is
    // issued immediately even for unsettled deliveries.
    //
    if (moved && link->link_direction == QD_INCOMING &&
        link->link_type != QD_LINK_ROUTER && !link->connected_link)
        qdr_link_issue_credit_CT(core, link, 1, false);

    return moved;
}

static void qdr_link_flow_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
    if (discard)
        return;

    qdr_link_t *link   = action->args.connection.link;
    int  credit        = action->args.connection.credit;
    bool drain         = action->args.connection.drain;
    bool activate      = false;
    bool drain_was_set = !link->drain_mode && drain;

    link->drain_mode = drain;

    //
    // If this is an attach-routed link, propagate the flow data downrange.
    // Note that the credit value is incremental.
    //
    if (link->connected_link) {
        qdr_link_t *clink = link->connected_link;

        if (clink->link_direction == QD_INCOMING)
            qdr_link_issue_credit_CT(core, link->connected_link, credit, drain);
        else {
            sys_mutex_lock(clink->conn->work_lock);
            qdr_add_link_ref(&clink->conn->links_with_deliveries, clink, QDR_LINK_LIST_CLASS_DELIVERY);
            sys_mutex_unlock(clink->conn->work_lock);
            qdr_connection_activate_CT(core, clink->conn);
        }

        return;
    }

    //
    // Handle the replenishing of credit outbound
    //
    if (link->link_direction == QD_OUTGOING && (credit > 0 || drain_was_set)) {
        sys_mutex_lock(link->conn->work_lock);
        if (DEQ_SIZE(link->undelivered) > 0 || drain_was_set) {
            qdr_add_link_ref(&link->conn->links_with_deliveries, link, QDR_LINK_LIST_CLASS_DELIVERY);
            activate = true;
        }
        sys_mutex_unlock(link->conn->work_lock);
    }

    //
    // Activate the connection if we have deliveries to send or drain mode was set.
    //
    if (activate)
        qdr_connection_activate_CT(core, link->conn);
}

static void qdr_link_check_credit_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
    if (discard)
        return;

    qdr_link_t *link = action->args.connection.link;
    qdr_link_issue_credit_CT(core, link, 0, false);
}

/**
 * Return the number of outbound paths to destinations that this address has.
 * Note that even if there are more than zero paths, the destination still may
 * be unreachable (e.g. an rnode next hop with no link).
 */
static long qdr_addr_path_count_CT(qdr_address_t *addr)
{
    return (long) DEQ_SIZE(addr->subscriptions) + (long) DEQ_SIZE(addr->rlinks) +
        (long) qd_bitmask_cardinality(addr->rnodes);
}

static void qdr_link_forward_CT(qdr_core_t *core, qdr_link_t *link, qdr_delivery_t *dlv, qdr_address_t *addr)
{
    if (addr && addr == link->owning_addr && qdr_addr_path_count_CT(addr) == 0) {
        //
        // We are trying to forward a delivery on an address that has no outbound paths
        // AND the incoming link is targeted (not anonymous).  In this case, we must put
        // the delivery on the incoming link's undelivered list.  Note that it is safe
        // to do this because the undelivered list will be flushed once the number of
        // paths transitions from zero to one.
        //
        // Use the action-reference as the reference for undelivered rather
        // than decrementing and incrementing the delivery ref_count.
        //
        DEQ_INSERT_TAIL(link->undelivered, dlv);
        dlv->where = QDR_DELIVERY_IN_UNDELIVERED;
        return;
    }

    int fanout = 0;

    if (addr) {
        fanout = qdr_forward_message_CT(core, addr, dlv->msg, dlv, false, link->link_type == QD_LINK_CONTROL);
        if (link->link_type != QD_LINK_CONTROL && link->link_type != QD_LINK_ROUTER)
            addr->deliveries_ingress++;
        link->total_deliveries++;
    }

    if (fanout == 0) {
        //
        // Message was not delivered, drop the delivery.
        //
        // If the delivery is not settled, release it.
        //
        if (!dlv->settled)
            qdr_delivery_release_CT(core, dlv);
        qdr_delivery_decref(dlv);
        qdr_link_issue_credit_CT(core, link, 1, false);
    } else if (fanout > 0) {
        if (dlv->settled) {
            //
            // The delivery is settled.  Keep it off the unsettled list and issue
            // replacement credit for it now.
            //
            qdr_link_issue_credit_CT(core, link, 1, false);

            //
            // If the delivery has no more references, free it now.
            //
            assert(!dlv->peer);
            qdr_delivery_decref(dlv);
        } else {
            //
            // Again, don't bother decrementing then incrementing the ref_count
            //
            DEQ_INSERT_TAIL(link->unsettled, dlv);
            dlv->where = QDR_DELIVERY_IN_UNSETTLED;

            //
            // If the delivery was received on an inter-router link, issue the credit
            // now.  We don't want to tie inter-router link flow control to unsettled
            // deliveries because it increases the risk of credit starvation if there
            // are many addresses sharing the link.
            //
            if (link->link_type == QD_LINK_ROUTER)
                qdr_link_issue_credit_CT(core, link, 1, false);
        }
    }
}

static void qdr_link_deliver_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
    if (discard)
        return;

    qdr_delivery_t *dlv  = action->args.connection.delivery;
    qdr_link_t     *link = dlv->link;

    //
    // If this is an attach-routed link, put the delivery directly onto the peer link
    //
    if (link->connected_link) {
        qdr_delivery_t *peer = qdr_forward_new_delivery_CT(core, dlv, link->connected_link, dlv->msg);

        //
        // Copy the delivery tag.  For link-routing, the delivery tag must be preserved.
        //
        peer->tag_length = action->args.connection.tag_length;
        memcpy(peer->tag, action->args.connection.tag, peer->tag_length);

        qdr_forward_deliver_CT(core, link->connected_link, peer);
        qd_message_free(dlv->msg);
        dlv->msg = 0;
        link->total_deliveries++;
        if (!dlv->settled) {
            DEQ_INSERT_TAIL(link->unsettled, dlv);
            dlv->where = QDR_DELIVERY_IN_UNSETTLED;

            //
            // Note, in this case the ref_count is left unchanged as we are transferring
            // the action's reference to the unsettled list's reference.
            //
        } else {
            //
            // If the delivery is settled, decrement the ref_count on the delivery.
            // This count was the owned-by-action count.
            //
            qdr_delivery_decref(dlv);
        }
        return;
    }

    //
    // NOTE: The link->undelivered list does not need to be protected by the
    //       connection's work lock for incoming links.  This protection is only
    //       needed for outgoing links.
    //

    if (DEQ_IS_EMPTY(link->undelivered)) {
        qdr_address_t *addr = link->owning_addr;
        if (!addr && dlv->to_addr)
            qd_hash_retrieve(core->addr_hash, dlv->to_addr, (void**) &addr);

        //
        // Give the action reference to the qdr_link_forward function.
        //
        qdr_link_forward_CT(core, link, dlv, addr);
    } else {
        //
        // Take the action reference and use it for undelivered.  Don't decref/incref.
        //
        DEQ_INSERT_TAIL(link->undelivered, dlv);
        dlv->where = QDR_DELIVERY_IN_UNDELIVERED;
    }
}

static void qdr_send_to_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
    qdr_field_t  *addr_field = action->args.io.address;
    qd_message_t *msg        = action->args.io.message;

    if (!discard) {
        qdr_address_t *addr = 0;

        qd_iterator_reset_view(addr_field->iterator, ITER_VIEW_ADDRESS_HASH);
        qd_hash_retrieve(core->addr_hash, addr_field->iterator, (void**) &addr);
        if (addr) {
            //
            // Forward the message.  We don't care what the fanout count is.
            //
            (void) qdr_forward_message_CT(core, addr, msg, 0, action->args.io.exclude_inprocess,
                                          action->args.io.control);
            addr->deliveries_from_container++;
        } else
            qd_log(core->log, QD_LOG_DEBUG, "In-process send to an unknown address");
    }

    qdr_field_free(addr_field);
    qd_message_free(msg);
}

static void qdr_update_delivery_CT(qdr_core_t *core, qdr_action_t *action, bool discard)
{
    qdr_delivery_t *dlv        = action->args.delivery.delivery;
    qdr_delivery_t *peer       = dlv->peer;
    bool            push       = false;
    bool            peer_moved = false;
    bool            dlv_moved  = false;
    uint64_t        disp       = action->args.delivery.disposition;
    bool            settled    = action->args.delivery.settled;

    //
    // Logic:
    //
    // If disposition has changed and there is a peer link, set the disposition of the peer
    // If settled, the delivery must be unlinked and freed.
    // If settled and there is a peer, the peer shall be settled and unlinked.  It shall not
    //   be freed until the connection-side thread settles the PN delivery.
    //
    if (disp != dlv->disposition) {
        //
        // Disposition has changed, propagate the change to the peer delivery.
        //
        dlv->disposition = disp;
        if (peer) {
            peer->disposition = disp;
            push = true;
        }
    }

    if (settled) {
        if (peer) {
            peer->settled = true;
            peer->peer = 0;
            dlv->peer  = 0;

            qdr_delivery_decref(dlv);
            qdr_delivery_decref(peer);

            if (peer->link) {
                peer_moved = qdr_delivery_settled_CT(core, peer);
                if (peer_moved)
                    push = true;
            }
        }

        if (dlv->link)
            dlv_moved = qdr_delivery_settled_CT(core, dlv);
    }

    if (push)
        qdr_delivery_push_CT(core, peer);

    //
    // Release the action reference, possibly freeing the delivery
    //
    qdr_delivery_decref(dlv);

    //
    // Release the unsettled references if the deliveries were moved
    //
    if (dlv_moved)
        qdr_delivery_decref(dlv);
    if (peer_moved)
        qdr_delivery_decref(peer);
}

/**
 * Check the link's accumulated credit.  If the credit given to the connection thread
 * has been issued to Proton, provide the next batch of credit to the connection thread.
 */
void qdr_link_issue_credit_CT(qdr_core_t *core, qdr_link_t *link, int credit, bool drain)
{
    bool drain_changed = link->drain_mode |= drain;
    bool activate      = drain_changed;

    link->drain_mode = drain;
    link->drain_mode_changed = drain_changed;

    link->incremental_credit_CT += credit;
    link->flow_started = true;

    if (link->incremental_credit_CT && link->incremental_credit == 0) {
        //
        // Move the credit from the core-thread value to the connection-thread value.
        //
        link->incremental_credit    = link->incremental_credit_CT;
        link->incremental_credit_CT = 0;
        activate = true;
    }

    if (activate) {
        //
        // Put this link on the connection's has-credit list.
        //
        sys_mutex_lock(link->conn->work_lock);
        qdr_add_link_ref(&link->conn->links_with_credit, link, QDR_LINK_LIST_CLASS_FLOW);
        sys_mutex_unlock(link->conn->work_lock);

        //
        // Activate the connection
        //
        qdr_connection_activate_CT(core, link->conn);
    }
}

/**
 * This function should be called after adding a new destination (subscription, local link,
 * or remote node) to an address.  If this address now has exactly one destination (i.e. it
 * transitioned from unreachable to reachable), make sure any unstarted in-links are issued
 * initial credit.
 *
 * Also, check the inlinks to see if there are undelivered messages.  If so, drain them to
 * the forwarder.
 */
void qdr_addr_start_inlinks_CT(qdr_core_t *core, qdr_address_t *addr)
{
    //
    // If there aren't any inlinks, there's no point in proceeding.
    //
    if (DEQ_SIZE(addr->inlinks) == 0)
        return;

    if (qdr_addr_path_count_CT(addr) == 1) {
        qdr_link_ref_t *ref = DEQ_HEAD(addr->inlinks);
        while (ref) {
            qdr_link_t *link = ref->link;

            //
            // Issue credit to stalled links
            //
            if (!link->flow_started)
                qdr_link_issue_credit_CT(core, link, link->capacity, false);

            //
            // Drain undelivered deliveries via the forwarder
            //
            if (DEQ_SIZE(link->undelivered) > 0) {
                //
                // Move all the undelivered to a local list in case not all can be delivered.
                // We don't want to loop here forever putting the same messages on the undelivered
                // list.
                //
                qdr_delivery_list_t deliveries;
                DEQ_MOVE(link->undelivered, deliveries);

                qdr_delivery_t *dlv = DEQ_HEAD(deliveries);
                while (dlv) {
                    DEQ_REMOVE_HEAD(deliveries);
                    qdr_link_forward_CT(core, link, dlv, addr);
                    dlv = DEQ_HEAD(deliveries);
                }
            }

            ref = DEQ_NEXT(ref);
        }
    }
}

void qdr_delivery_push_CT(qdr_core_t *core, qdr_delivery_t *dlv)
{
    if (!dlv || !dlv->link)
        return;

    qdr_link_t *link = dlv->link;
    bool activate = false;

    sys_mutex_lock(link->conn->work_lock);
    if (dlv->where != QDR_DELIVERY_IN_UNDELIVERED) {
        qdr_delivery_incref(dlv);
        qdr_add_delivery_ref(&link->updated_deliveries, dlv);
        qdr_add_link_ref(&link->conn->links_with_deliveries, link, QDR_LINK_LIST_CLASS_DELIVERY);
        activate = true;
    }
    sys_mutex_unlock(link->conn->work_lock);

    //
    // Activate the connection
    //
    if (activate)
        qdr_connection_activate_CT(core, link->conn);
}