text:
title: CeilometerQuickStart
category: documentation
authors: eglynn, jpichon
wiki_category: Documentation
wiki_title: CeilometerQuickStart
wiki_revision_count: 43
wiki_last_updated: 2014-11-26
---

__NOTOC__

# Ceilometer Quickstart

This guide is intended to get you up and running quickly with Ceilometer:

*   metering cloud resource usage
*   understanding the foundational Ceilometer concepts and terminology
*   interacting with the aggregation API via the command line interface
*   understanding how the metering store is structured
*   alarming on resource performance

*This guide is predicated on at least the Havana final version of the packages being used, i.e. openstack-ceilometer-2013.2-1 or later (as the service start-up instructions for older versions of the packages from the H cycle would differ somewhat).*

### Prerequisites

The general prerequisites are identical to the [Havana QuickStart](QuickStartLatest), and in fact if you follow that guide right now you will end up with Ceilometer installed by default once your `packstack` run completes .

However, if you are installing on a resource-constrained VM, some prior setup can make your life easier. The default & most feature-complete storage driver used by Ceilometer is `mongodb`, which on installation eagerly pre-allocates large journal files *etc*. This is not normally an issue, but the default service timeout of 90 seconds imposed by systemd *may* not suffice for the pre-allocation to complete on resource-starved VMs. However this can be easily worked around in that scenarion by pre-installing mongo with a modified startup timeout, as follows:

         sudo yum install -y mongodb-server mongodb
         sudo sed -i '/^`\[Service\]`$/ a\
           TimeoutStartSec=360' /usr/lib/systemd/system/mongod.service
         sudo service mongod start
         sudo service mongod status
         sudo service mongod stop

### Verification

Once your `packstack` run is complete, you're probably eager to verify that Ceilometer is properly installed and working it should.

Before we do that, a few words on how Ceilometer is realized as a set of agents and services. It comprises at least four separate daemons, each with a specific function in the metering pipeline:

*   **compute** agent: polls the local libvirt daemon to acquire performance data for the local instances, messages and emits these data as AMQP notifications
*   **central** agent: polls the public RESTful APIs of other openstack services such as nova and glance, in order to keep tabs on resource existence
*   **collector** service: consumes AMQP notifications from the agents and other openstack services, then dispatch these data to the metering store
*   **API** service: presents aggregated metering data to consumers (such as billing engines, analytics tools *etc*.)
*   **alarm-evaluator** service: determines when alarms fire due to the associated statistic trend crossing a threshold over a sliding time window
*   **alarm-notifier** service: initiates alarm actions, for example calling out to a webhook with a description of the alarm state transition

 In a `packstack` "all in one" installation, all of these services will be running on your single node. In a wider deployment, the main location constraint is that the compute agent is required to run on all nova compute nodes. Assuming "all in one" for now, check that all services are running smoothly:

       export CEILO_SVCS='compute central collector api alarm-evaluator alarm-notifier'
       for svc in $CEILO_SVCS ; do sudo service openstack-ceilometer-$svc status ; done

For your peace of mind, ensure that there are no errors in the Ceilometer logs at this time:

       for svc in $CEILO_SVCS ; do sudo grep ERROR /var/log/ceilometer/${svc}.log ; done

### Basic Concepts

Getting up to speed with Ceilometer involves getting to grips a few basic concepts and terms.

#### Meters

Meters simply measure a particular aspect of resource usage (e.g. the existence of a running instance) or of ongoing performance (e.g. the current CPU utilization % for that instance). As such meters exist per-resource, in that there is a separate `cpu_util` meter for example for each instance. The lifecycle of meters is also decoupled from the existence of the related resources, in the sense that the meter continues to exist *after* the resource has been terminated. While that may seem strange initially, think about how otherwise you could avoid being billed simply by shutting down all your instances the day before your cloud provider kicks off their monthly billing run!

All meters have a string name, a unit of measurement, and a type indicating whether values are monotonically increasing (`cumulative`), interpreted as a change from the previous value (`delta`), or a standalone value relating only to the current duration (`gauge`).

In earlier iterations of Ceilometer, we often used 'counter' as a synonym for 'meter', and this usage though now deprecated persists in some older documentation and deprecated aspects of the command line interpreter.

#### Samples

Sample are simply individual datapoints associated with a particular meter. As such, all samples encompass the same attributes as the meter itself, but with the addition of a timestamp and and a value (otherwise known as the sample 'volume').

#### Statistics

If a sample is a single datapoint, then a statistic is a set of such datapoints aggregates over a time duration. Ceilometer currently employs 5 different aggregation functions:

*   **count**: the number of samples in each period
*   **max**: the maxima of the sample volumes in each period
*   **min**: the minima of the sample volumes in each period
*   **avg**: the average of the sample volumes over each period
*   **sum**: the sum of the sample volumes over each period

 Note that *all* of these aggregation functions are applied for every statistic calculated. This may seem wasteful if you're only interested in one of the values, but in practice hardly any extra computation cost is incurred due to the map-reduce scheme used to calculate these values.

Also there is some potential confusion in there being both a duration *and* a period associated with these statistics. The duration is simply the overall time-span over which a single query applies, whereas the period is the time-slice length into which this duration is divided for aggregation purposes. So for example, if I was interested in the hourly average CPU utilization over a day, I would provide midnight-to-midnight start and end timestamps on my query giving a duration of 24 hours, while also specifying a period of 3600 seconds to indicate that the finegrained samples should be aggregated over each hour within that day.

#### Pipelines

Pipelines are composed of a metering data source that produces certain enumerated or wildcarded meters at a certain cadence, which are fed through a chain of zero or more transformers to massage the data in various ways, before being emitted to the collector via a publisher.

Example of transformers shipped with Ceilometer include:

*   **unit_conversion**: apply a scaling conversion to use a different unit than originally supplied with the observed data, for example converting a temperature from °F to °C (the scaling rule is configurable, so that any reasonable unit conversion expressible in python may be implemented)
*   **rate_of_change**: derives a secondary meter from directly observed data based on the calculated rate of change by sampling the sequence of datapoint, with an associated scaling rule
*   **accumulator**: gather several datapoints before emitting in a batch

 Multiple publishers are also supported, including:

*   **rpc://**: emit metering data for collector over AMQP
*   **<udp://>**: emit metering data for collector over lossy UDP (useful for metric data collected for alarming purposes, but not suitable for metering data to feed into a billing system due to the obvious retention requirements)
*   **<file://>**: emit metering data into a file

These pipelines are configured via a YAML file which is explained in detail below.

#### Alarms

Alarms are a new feature in Ceilometer for Havana intended to provide user-oriented Monitoring-as-a-Service for Openstack, with Heat autoscaling being the main motivating use-case, but also having general purpose utility. Essentially an alarm is just a set of rules defining a monitor, plus a current state, with edge-triggered actions associated with target states. These alarms follow a tri-state model of `ok`, `alarm`, and `insufficient data`.

For conventional threshold-oriented alarms, state transitions are governed by:

*   a *static* threshold value & comparison operator
*   against which a selected meter statistic is compared
*   over an evaluation window of configurable length into the recent past.

We also support the concept of a meta-alarm, which aggregates over the current state of a set of other basic alarms combined via a logical operator (AND/OR).

A key associated concept is the notion of *dimensioning* which defines the set of matching meters that feed into an alarm evaluation. Recall that meters are per-resource-instance, so in the simplest case an alarm might be defined over a particular meter applied to *all* resources visible to a particular user. More useful however would the option to explicitly select which specific resources we're interested in alarming on. On one extreme we would have narrowly dimensioned alarms where this selection would have only a single target (identified by resource ID). On the other extreme, we'd have widely dimensioned alarms where this selection identifies many resources over which the statistic is aggregated, for example all instances booted from a particular image or all instances with matching user metadata (the latter is how Heat identifies autoscaling groups).

### Configuration

The shipped Ceilometer configuration is intended to be usable out-of-the-box. However, there are a few tweaks you may want to make while exploring Ceilometer functionality.

#### Pipeline configuration

The pipeline definitions are read by default from `/etc/ceilometer/pipeline.yaml` though this location may be overridden via the `pipeline_cfg_file` config option (for example to allow different ceilometer services use different pipelines.

The most likely pipeline config elements you might want to experiment initially initially would be:

*   **`interval`**: defines the cadence of data acquisition by controlling the polling period, which default to 600 seconds (10 minutes).
*   **`meters`**: a list of meters that the current pipeline applies to, either explicitly enumerated with negation via `!` or wildcarded.
*   **`transformers`**: a list of named transformers and their parameters, to be loaded as `stevedore` extensions.

 To become more familiar with the possibilities offered by this configuration file, let's examine the shipped `pipeline.yaml`:

       ---
       -
           name: meter_pipeline
           interval: 600
           meters:
               - "*"
           transformers:
           publishers:
               - rpc://
       -
           name: cpu_pipeline
           interval: 600
           meters:
               - "cpu"
           transformers:
               - name: "rate_of_change"
                 parameters:
                     target:
                         name: "cpu_util"
                         unit: "%"
                         type: "gauge"
                         scale: "100.0 / (10**9 * (resource_metadata.cpu_number or 1))"
           publishers:
               - rpc://

This file defines two separate pipeline, named `meter_pipeline` and `cpu_pipeline`,

The first is very straight-forward, running at the default cadence of 600s and applying to all primary meters (`meters: - "*"`) which are emitted unchanged over AMQP.

The second is more interesting, applying to only the `cpu` meter, transforming this via the `rate_of_change` transformer into the derived `cpu_util` meter. The effect here is to transformer the primary observation of cumulative CPU time in nanosecond into a gauge value as a percentage of the notional maximum total CPU time over that preceding duration scaled by the number of vCPUs allocated to the instance. The scaling rule is expressed as a fragment of python which handles the percentage conversion, nanosecond scaling and taking the number of CPUs into account, all defined very concisely in configuration.

An example modification would be something like increasing the cadence of `cpu_util` from once per 10 minutes to once a minute:

       sudo sed -i '/^ *name: cpu_pipeline$/ { n ; s/interval: 600$/interval: 60/ }' /etc/ceilometer/pipeline.yaml
       sudo service openstack-ceilometer-compute restart

Note that we only need to restart the `compute` and not the `central` even though both share the same pipeline config by default, because the particular meter impacted by the change is only gathered by the former agent.

#### Service configuration

The service configuration is read by default from two sources, via the same pattern as you've encountered with the other openstack services in RDO:

*   **distribution config**: `/usr/share/ceilometer/ceilometer-dist.conf` containing the distro-specific overrides over upstream defaults
*   **user-editable config**: `/etc/ceilometer/ceilometer.conf` containing commented-out setting for the all configuration options with help strings and default values

 If you wish to change some configuration option, the latter file is one to edit. This is laid out in the familiar sectioned format as provided by the Olso config library, with configuration options under `[DEFAULT]`, `[database]`, `[api]`, `[alarm]` *etc*.

As always, you can choose to manually edit this file or else use the convenient `openstack-config` utility from the `openstack-utils` package, for example:

       sudo openstack-config --set /etc/ceilometer/ceilometer.conf DEFAULT debug true

to set the logging level to debug as opposed to the default warning. As always, services must be restarted for config changes to take effect.

### Exploring with the CLI

First ensure that the latest version of CLI package is installed:

       sudo rpm -qa | awk -F- '/python-ceilometerclient/ {print $3}'
       1.0.8

which will allow you to access the latest API additions, for example alarm history and the separation of alarm representation from the encapsulated rules.

We will proceed to explore each of the basic concepts described earlier in this guide. But before we do so, it's worth mentioning a common conceptual banana skin that often confuses new Ceilometer users at this early stage. Recall that metering is all about measuring user-visible cloud resource usage - if there ain't any user-visible resources in your cloud, no metering data will be generated. So let's start by ensuring some resources are actually present, firstly the basic building blocks, some VM images in `glance`:

       glance image-list

If this list empty, let's quickly grab a small basic `cirros` image that we'll later use to spin up some instances:

       sudo yum install -y wget
` wget `[`http://launchpad.net/cirros/trunk/0.3.0/+download/cirros-0.3.0-x86_64-uec.tar.gz`](http://launchpad.net/cirros/trunk/0.3.0/+download/cirros-0.3.0-x86_64-uec.tar.gz)
       tar zxvf cirros-0.3.0-x86_64-uec.tar.gz
       glance image-create --name cirros-aki --is-public True --container-format aki --disk-format aki \
         --file cirros-0.3.0-x86_64-vmlinuz
       glance image-create --name cirros-ari --is-public True --container-format ari --disk-format ari \
         --file cirros-0.3.0-x86_64-initrd
       glance image-create --name cirros-ami --is-public True --container-format ami --disk-format ami \
         --property kernel_id=$(glance image-list | awk '/cirros-aki/ {print $2}') \
         --property ramdisk_id=$(glance image-list | awk '/cirros-ari/ {print $2}') --file cirros-0.3.0-x86_64-blank.img

Then spin up an instance booted from that image:

       IMAGE_ID=$(glance image-list | awk '/cirros-ami/ {print $2}')
       nova boot --image $IMAGE_ID --flavor 1 test_instance

Wait for that instance to become active and we're good to go!

       watch 'nova show test_instance'

#### Displaying meters

Individual meters are displayed via the CLI `meter-list` command:

       $ ceilometer meter-list
       +----------------------------+------------+-----------+---------------+-----------+--------------+
       | Name                       | Type       | Unit      | Resource ID   | User ID   | Project ID   |
       +----------------------------+------------+-----------+---------------+-----------+--------------+
       | cpu                        | cumulative | ns        | INSTANCE_ID_1 | USER_ID_A | PROJECT_ID_X |
       | cpu                        | cumulative | ns        | INSTANCE_ID_2 | USER_ID_B | PROJECT_ID_Y |
       | cpu                        | cumulative | ns        | INSTANCE_ID_3 | USER_ID_C | PROJECT_ID_Z |
       | cpu_util                   | gauge      | %         | INSTANCE_ID_1 | USER_ID_A | PROJECT_ID_X |
       | cpu_util                   | gauge      | %         | INSTANCE_ID_3 | USER_ID_C | PROJECT_ID_Z |
       | disk.ephemeral.size        | gauge      | GB        | INSTANCE_ID_1 | USER_ID_A | PROJECT_ID_X |
       | disk.ephemeral.size        | gauge      | GB        | INSTANCE_ID_2 | USER_ID_B | PROJECT_ID_Y |
       | disk.ephemeral.size        | gauge      | GB        | INSTANCE_ID_3 | USER_ID_C | PROJECT_ID_Z |
       | ... [snip]                                                                                |
       +----------------------------+------------+-----------+---------------+-----------+--------------+

As you can see in the example output above, all meters are listed for all resources that existed since metering began (modulo data expiry if configured). We can use the `--query` option to limit the output to a specific resource, project or user for example.

       $ ceilometer meter-list --query project=PROJECT_ID_Y;user=USER_ID_B
       +----------------------------+------------+-----------+---------------+-----------+--------------+
       | Name                       | Type       | Unit      | Resource ID   | User ID   | Project ID   |
       +----------------------------+------------+-----------+---------------+-----------+--------------+
       | cpu                        | cumulative | ns        | INSTANCE_ID_2 | USER_ID_B | PROJECT_ID_Y |
       | disk.ephemeral.size        | gauge      | GB        | INSTANCE_ID_2 | USER_ID_B | PROJECT_ID_Y |
       | ... [snip]                                                                                |
       +----------------------------+------------+-----------+---------------+-----------+--------------+

The syntax of that `--query` (or `-q`) option is common across several CLI commands target'ing the v2 API, the syntax being:

` -q `<field1><operator1><value1>`;`<field2><operator2><value2>`;...;`<field_n><operator_n><value_n>

which are translated by the CLI to a sequence of [WSME](//pypi.python.org/pypi/WSME) query parameters.

#### Displaying datapoints

Individual datapoints for a particular meter name are displayed via the CLI `samples-list` command:

       $ ceilometer sample-list --meter cpu
       +---------------+------+------------+---------------+------+---------------------+
       | Resource ID   | Name | Type       | Volume        | Unit | Timestamp           |
       +---------------+------+------------+---------------+------+---------------------+
       | INSTANCE_ID_1 | cpu  | cumulative | 1.6844e+11    | ns   | 2013-10-01T08:48:29 |
       | INSTANCE_ID_1 | cpu  | cumulative | 1.7039e+11    | ns   | 2013-10-01T08:58:28 |
       | INSTANCE_ID_1 | cpu  | cumulative | 1.7234e+11    | ns   | 2013-10-01T09:08:28 |
       | INSTANCE_ID_1 | cpu  | cumulative | 1.743e+11     | ns   | 2013-10-01T09:18:28 |
       | INSTANCE_ID_1 | cpu  | cumulative | 1.7626e+11    | ns   | 2013-10-01T09:28:28 |
       | ... [snip]                                                                     |
       | INSTANCE_ID_2 | cpu  | cumulative | 2.9833e+11    | ns   | 2013-10-01T08:48:29 |
       | INSTANCE_ID_2 | cpu  | cumulative | 2.6028e+11    | ns   | 2013-10-01T08:58:28 |
       | INSTANCE_ID_2 | cpu  | cumulative | 3.7156e+11    | ns   | 2013-10-01T09:08:28 |
       | INSTANCE_ID_2 | cpu  | cumulative | 3.7987e+11    | ns   | 2013-10-01T09:18:28 |
       | INSTANCE_ID_2 | cpu  | cumulative | 2.6555e+11    | ns   | 2013-10-01T09:28:28 |
       | ... [snip]                                                                     |
       +---------------+------+------------+---------------+------+---------------------+

Note that the samples relate to multiple resources (assuming more than one instances was spun up in this case) and are grouped by resource ID, and sorted by timestamp. Since the query applies to this meter name as it pertains to *all* resources, if metering has been running for any reasonable duration, this command unadorned can turn into a bit of a firehose in terms the sheer volume of data returned. As before, we can rely on the `-q` option to constrain the query, for example by resource id and timestamp:

       $ ceilometer sample-list --meter cpu -q 'resource_id=INSTANCE_ID_1;timestamp>2013-10-01T09:00:00;timestamp<=2013-10-01T09:30:00'
       +---------------+------+------------+---------------+------+---------------------+
       | Resource ID   | Name | Type       | Volume        | Unit | Timestamp           |
       +---------------+------+------------+---------------+------+---------------------+
       | INSTANCE_ID_1 | cpu  | cumulative | 1.7234e+11    | ns   | 2013-10-01T09:08:28 |
       | INSTANCE_ID_1 | cpu  | cumulative | 1.743e+11     | ns   | 2013-10-01T09:18:28 |
       | INSTANCE_ID_1 | cpu  | cumulative | 1.7626e+11    | ns   | 2013-10-01T09:28:28 |
       +---------------+------+------------+---------------+------+---------------------+

to restrict the query to samples for a particular instance that occurred within the specified half hour time window,

#### Aggregating statistics

Individual datapoints for a particular meter may be aggregated into consolidated statistics via the CLI `statistics` command:

       $ ceilometer statistics --meter cpu_util
       +--------+--------------+------------+-------+------+-----+-----+-----+----------+----------------+----
       | Period | Period Start | Period End | Count | Min  | Max | Sum | Avg | Duration | Duration Start | ...
       +--------+--------------+------------+-------+------+-----+-----+-----+----------+----------------+----
       | 0      | PERIOD_START | PERIOD_END | 2024  | 0.25 | 6.2 | 550 | 2.9 | 85196.0  | DURATION_START | ...
       +--------+--------------+------------+-------+------+-----+-----+-----+----------+----------------+----

(output is narrowed for brevity here). The thing to notice here is that by default *all* samples for *all* meters matching the given name are aggregated for *all* time. It would be more normal to require that:

1.  the samples feeding into the statistics are constrained by resource or some other attribute
2.  the overall duration is bounded by start and end timestamp
3.  this bounded duration is further subdivided into timeslices

 These extra constraints may all be expressed on the command line:

       $ ceilometer --debug statistics -m cpu_util -q 'timestamp>START;timestamp<=END' --period 60
       +--------+--------------+------------+-------+-----+-----+-----+-----+----------+----------------+----
       | Period | Period Start | Period End | Count | Min | Max | Sum | Avg | Duration | Duration Start | ...
       +--------+--------------+------------+-------+-----+-----+-----+-----+----------+----------------+----
       | 60     | START        | START+60   | 2     | 1.5 | 2.5 | 4.0 | 2.0 | 0.0      | DURATION_START | ...
       | 60     | START+60     | START+120  | 2     | 2.5 | 3.5 | 6.0 | 3.0 | 0.0      | DURATION_START | ...
       | ...[snip]
       +--------+--------------+------------+-------+-----+-----+-----+-----+----------+----------------+----

#### Using alarms

Before creating any alarms, ensure that the relevant Ceilometer alarming services are running:

       export CEILO_ALARM_SVCS='evaluator notifier'
       for svc in $CEILO_ALARM_SVCS; do sudo service openstack-ceilometer-alarm-$svc status; done

An example of creating a threshold-oriented alarm, based on a upper bound on the CPU utilization for a particular instance:

       $ ceilometer alarm-threshold-create --name cpu_high --description 'instance running hot'  \
         --meter-name cpu_util  --threshold 70.0 --comparison-operator gt  --statistic avg \
         --period 600 --evaluation-periods 3 \
         --alarm-action 'log://' \
         --query resource_id=INSTANCE_ID
       +---------------------------+-----------------------------------------------------+
       | Property                  | Value                                               |
       +---------------------------+-----------------------------------------------------+
       | meter_name                | cpu_util                                            |
       | alarm_actions             | [u'log://']                                         |
       | user_id                   | USER_ID                                             |
       | name                      | cpu_high                                            |
       | evaluation_periods        | 3                                                   |
       | statistic                 | avg                                                 |
       | enabled                   | True                                                |
       | period                    | 600                                                 |
       | alarm_id                  | ALARM_ID                                            |
       | state                     | insufficient data                                   |
       | query                     | resource_id == INSTANCE_ID                          |
       | insufficient_data_actions | []                                                  |
       | repeat_actions            | False                                               |
       | threshold                 | 70.0                                                |
       | ok_actions                | []                                                  |
       | project_id                | PROJECT_ID                                          |
       | type                      | threshold                                           |
       | comparison_operator       | gt                                                  |
       | description               | instance running hot                                |
       +---------------------------+-----------------------------------------------------+

This creates an alarm that will fire when the average CPU utilization for an individual instance exceeds 70% for three consecutive 10 minute periods. The notification is this case is simply a log message, though it could alternatively be a webhook URL.

You can display all your alarms via:

       $ ceilometer alarm-list
       +----------+----------+-------------------+---------+------------+---------------------------------+
       | Alarm ID | Name     | State             | Enabled | Continuous | Alarm condition                 |
       +----------+----------+-------------------+---------+------------+---------------------------------+
       | ALARM_ID | cpu_high | insufficient data | True    | False      | cpu_util > 70.0 during 3 x 600s |
       +----------+----------+-------------------+---------+------------+---------------------------------+

In this case, the state is reported as `insufficient data` which could indicate that:

*   metrics have not yet been gathered about this instance over the evaluation window into the recent past (e.g. a brand-new instance)
*   or, that the identified instance is not visible to the user/tenant owning the alarm
*   or, simply that an alarm evaluation cycle hasn't kicked off since the alarm was created (by default, alarms are evaluated once per minute).

 Once the state of the alarm has settled down, we might decide that we set that bar too low with 70%, in which case the threshold (or most any other alarm attribute) can be updated as follows:

       $ ceilometer alarm-update --threshold 75 -a ALARM_ID
       +---------------------------+-----------------------------------------------------+
       | Property                  | Value                                               |
       +---------------------------+-----------------------------------------------------+
       | meter_name                | cpu_util                                            |
       | alarm_actions             | [u'log://']                                         |
       | user_id                   | USER_ID                                             |
       | name                      | cpu_high                                            |
       | evaluation_periods        | 3                                                   |
       | statistic                 | avg                                                 |
       | enabled                   | True                                                |
       | period                    | 600                                                 |
       | alarm_id                  | ALARM_ID                                            |
       | state                     | insufficient data                                   |
       | query                     | resource_id == INSTANCE_ID                          |
       | insufficient_data_actions | []                                                  |
       | repeat_actions            | False                                               |
       | threshold                 | 75.0                                                |
       | ok_actions                | []                                                  |
       | project_id                | PROJECT_ID                                          |
       | type                      | threshold                                           |
       | comparison_operator       | gt                                                  |
       | description               | instance running hot                                |
       +---------------------------+-----------------------------------------------------+

The change will take effect from the next evaluation cycle, which by default occurs every minute.

Over time the state of the alarm may change often, especially if the threshold is chosen to be close to the trending value of the statistic. We can follow the history of an alarm over its lifecycle via the audit API:

       $ ceilometer alarm-history -a ALARM_ID
       +------------------+----------------------------+---------------------------------------+
       | Type             | Timestamp                  | Detail                                |
       +------------------+----------------------------+---------------------------------------+
       | creation         | 2013-10-01T16:20:29.238000 | name: cpu_high                        |
       |                  |                            | description: instance running hot     |
       |                  |                            | type: threshold                       |
       |                  |                            | rule: cpu_util > 70.0 during 3 x 600s |
       | state transition | 2013-10-01T16:20:40.626000 | state: ok                             |
       | rule change      | 2013-10-01T16:22:40.718000 | rule: cpu_util > 75.0 during 3 x 600s |
       +------------------+----------------------------+---------------------------------------+

Finally, an alarm that's no longer required can be disabled:

      $ ceilometer alarm-update --enabled False -a ALARM_ID

or deleted permanently:

      $ ceilometer alarm-delete -a ALARM_ID

### Exploring the metering store

Ceilometer uses `mongodb` by default to store metering data, though alternative pluggable storage drivers are also provided for sqlalchemy, db2 and HBase. Only the `mongodb` storage driver is considered feature-complete at this time, so this is the recommended choice for production.

Before we begin to explore the datastore, ensure that the `mongo` client is installed locally:

       sudo yum install -y mongodb

The top-level structure can be seen by showing the available collections:

       $ mongo ceilometer
       MongoDB shell version: 2.4.6
       connecting to: ceilometer
       > show collections
       alarm
       alarm_history
       meter
       project
       resource
       system.indexes
       user

At the heart of the datastore is the `meter` collection containing the actual metering datapoints, and from which queries on meters, samples and statistics are satisfied. The `alarm` and `alarm_history` collections contain alarm rules & state and audit trails respectively. The `project` and `user` collections concern identity, referring to the known tenants and users respectively. Whereas the `resource` collection contains an entry per unique metered resource (instance, image, volume etc.), storing the metadata thereof and linking back to the related meters.

Note also the explicitly established system indices, which are created on demand by the storage driver:

       > query = {'name': {$ne: '_id_'}}             // only include explicitly named indices
       > projection = {'key': 1, 'ns': 1, 'name': 1} // project onto key, namespace, name
       > db.system.indexes.find(query, projection)
       { "key" : { "user_id" : 1, "source" : 1 }, "ns" : "ceilometer.resource", "name" : "resource_idx" }
       { "key" : { "resource_id" : 1, "user_id" : 1, "counter_name" : 1, "timestamp" : 1, "source" : 1 }, "ns" : "ceilometer.meter", "name" : "meter_idx" }
       { "key" : { "timestamp" : -1 }, "ns" : "ceilometer.meter", "name" : "timestamp_idx" }

The keys over which each index extends, in addition to the sort order (1 indicating ascending, -1 indicating decending) is revealed in the query result above. So for example, we see there's an index over the meter collection based on the timestamp attribute, ordered from most to least recent.

Unlike relational databases which have static schemata requiring careful management as they evolve, `mongo` is much more flexible and allows the structure of documents in a collection to change over time. Hence for this storage layer we do not have an analogue of the familiar `sqlalchemy-migrate` and/or `alembic` schema upgrade/downgrade scripts that are widely used across the OpenStack services. However there are several [tools available](http://skratchdot.com/projects/mongodb-schema) that allow a schema to be inferred from the observed document structure, if that would enhance your understanding of the store structure.

Now you could of course continue your exploration by looking at the raw documents stored in each of the Ceilometer collections, but these data are usually more conveniently retrieved via the API layer. However, there are cases were these data can be usefully processed directly, generally to aggregate in ways not currently supported by the Ceilometer API.

Say for example you wanted to see how much variance in CPU utilization there has been across the instances owned by a certain tenant. Casting your mind back to stats 101, you recall that such a question can be answered with the familiar concept of standard deviation. Now since standard deviation is not currently included in the set of statistical aggregates exposed in the v2 Ceilometer API, you could proceed to calculate it directly in `mongo` via map-reduce with some simple javascript:

       > function map() {
             emit(this.resource_id, {sum: this.counter_volume, count: 1, weighted_distances: 0});
         }
       > function reduce(key, mapped) {
             var merge = mapped[0];
             for (var i = 1 ; i < mapped.length ; i++) {
                 var deviance = (merge.sum / merge.count) - mapped[i].sum;
                 var weight = merge.count / ++merge.count;
                 merge.weighted_distances += (Math.pow(deviance, 2) * weight);
                 merge.sum += mapped[i].sum;
             }
             return merge;
         }
       > function complete(key, reduced) {
             reduced.stddev = Math.sqrt(reduced.weighted_distances / reduced.count);
             return reduced;
        }
       > db.meter.mapReduce(map,
                            reduce,
                            {finalize: complete,
                             out: {merge: 'cpu_util_deviation'},
                             query: {'counter_name': 'cpu_util',
                                     'project_id': PROJECT_ID}})
        ...
       > projection = { 'value.stddev': 1 }  // ignore partial results
       > db.cpu_util_deviation.find({}, projection)
       { "_id" : "INSTANCE_ID_1", "value" : { "stddev" : 0.030034533016630435 } }
       { "_id" : "INSTANCE_ID_2", "value" : { "stddev" : 0.3418399151135359 } }
       ...

We leave it as an exercise for the reader to compare the aggregate values directly calculated above in the partial results (sum, count) with those reported via the statistics API.

<Category:Documentation>


text:
package zmre.sbt

import sbt._
import sbt.TestResult.Value
import sbt.testing.Status._
import sbt.testing.Event
import scala.util.Try
//import org.scalatest.exceptions._

class UnitTestLogger(dir: File, srcFiles: => Seq[File], notify: () => Unit) extends TestsListener {
  import CompileLogger._

  val output = dir / "testissues.log"

  def clearLog() = {
    //println("Clearing test log")
    IO.delete(output)
    IO.touch(List(output))
  }

  def doInit() = clearLog()
  def doComplete(finalResult: TestResult.Value) = {
      notify()
  }

  def startGroup(name: String): Unit = {}

  def testEvent(events: TestEvent): Unit = {
    // TestEvent http://www.scala-sbt.org/0.13/api/index.html#sbt.TestEvent
    // has .result: Option[Value] and .detail[Seq[Event]]
    // Event, I think, is
    // http://grepcode.com/file/repo1.maven.org/maven2/org.scala-sbt/test-interface/1.0/sbt/testing/Event.java?av=f
    // event.throwable is OptionalThrowable, shown here:
    // http://grepcode.com/file/repo1.maven.org/maven2/org.scala-sbt/test-interface/1.0/sbt/testing/OptionalThrowable.java
    // status is an enum with these options: Success, Error, Failure, Skipped,
    // Ignored, Pending, Canceled
    // TODO: add a warning for Pending and Ignored tests?
    for {
      event <- events.detail if event.status == Failure || event.status == Error
      error <- Try(event.throwable.get)
      (file, line) <- findInStackTrace(error.getStackTrace)
    } { append(output, "error", file, line, error.getMessage); notify() }
  }

  def endGroup(name: String, t: Throwable): Unit = {}

  def endGroup(name: String, v: Value): Unit = {}

  def findInStackTrace(trace: Array[StackTraceElement]): Option[(File, Int)] =
    { for {
      elem <- trace
      file <- findSource(elem.getFileName)
    } yield (file, elem.getLineNumber) }.headOption

  def findSource(name: String): Option[File] = srcFiles find { file => file.getName endsWith name }
}

object UnitTestLogger {
  def apply(dir: File, srcFiles: Seq[File], notify: () => Unit): TestReportListener =
    new UnitTestLogger(dir, srcFiles, notify)
}

redpajama_set_name: RedPajamaGithub


text:
<?xml version="1.0" encoding="utf-8"?>
<resources>

    <style name="PreferenceFragmentList.Material">
        <item name="android:paddingStart">0dp</item>
        <item name="android:paddingEnd">0dp</item>
    </style>
</resources>


text:
<?php

namespace Markup\Test\TestCase\Highlighter;

use Cake\Core\Configure;
use Cake\TestSuite\TestCase;
use Markup\Highlighter\JsHighlighter;

class JsHighlighterTest extends TestCase {

	/**
	 * @var \Markup\Highlighter\JsHighlighter
	 */
	protected $highlighter;

	/**
	 * @return void
	 */
	public function setUp(): void {
		parent::setUp();

		Configure::write('Highlighter', [
				'debug' => false,
		]);

		$this->highlighter = new JsHighlighter();
	}

	/**
	 * @return void
	 */
	public function tearDown(): void {
		parent::tearDown();

		unset($this->highlighter);
	}

	/**
	 * @return void
	 */
	public function testConvert(): void {
		$result = $this->highlighter->highlight('My text');
		$expected = '
My text
';
		$this->assertSame($expected, $result);
	}

}


text:
(function () {
  'use strict'

  /**
   * @ngdoc service
   * @name app.service:nedb
   * @description
   * nedb - embeded no-SQL database engine
   */
  angular
    .module('app')
    .service('ExportService', ExportService)

  /* @ngInject */
  function ExportService ($q, DataService, _, json2csv, jszip, moment) {

    this.createCriteriaBlob = () => {
      return DataService.findAllCriteria().then(criteria => {
        const fileContent = JSON.stringify(criteria)
        const contentType = 'text/json'
        return new Blob([fileContent], { type: contentType })
      })
    }

    this.createDataForAnalysisBlob = () => {
      let promises = [ generateCriteriaCsv(), generateStudentGradesCsv() ]
      return $q.all(promises).then( ([criteriaCsv, studentGradesCsv]) => {
        // console.log(criteriaCsv);
        // console.log(studentGradesCsv);
        let zip = new jszip()
        zip.file('criteria.csv', criteriaCsv)
        zip.file('student-grades.csv', studentGradesCsv)

        return zip.generateAsync({type:"base64"}).then(base64 =>  {
          const blob = base64toBlob(base64, { type: 'application/zip' })
          return blob
        }, err => {
          console.log(err)
          // deferred.reject();
        })
      })
    }

    // TODO: this is duplicate from studentController.js
    let criterionSectionAsString = (section) => {
      const strList = []
      // drop trailing non-number elements (i.e. empty values)
      const section_trimmedRight = _.dropRightWhile(section, e => !_.isNumber(e))
      _.forEach(section_trimmedRight, e => { strList.push(e.toString()) })
      return strList.join('.')
    }

    let convertObjectToCsv = (obj) => {
      const deferred = $q.defer()
      json2csv(obj, (err, csv) => {
        if (err) {
          deferred.reject(err)
          // console.log(err);
        } else {
          deferred.resolve(csv)
        }
      });
      return deferred.promise;
    }

    let generateCriteriaCsv = () => {
      return DataService.findAllCriteria().then(criteria => {
        let data = []
        let counter = 1
        _.forEach(criteria, criterion => {
          data.push( {
            header : `Criterion${counter++}`,
            section : criterionSectionAsString(criterion.section),
            description : criterion.description,
            points : criterion.points
          })
        })
        // console.log(data);
        return convertObjectToCsv({data})
      })
    }

    let generateStudentGradesCsv = () => {
      const promises = [ DataService.findAllCriteria(), DataService.findAllStudents(), DataService.getAllGroups() ]
      return $q.all(promises).then(( [criteria, students, groups] ) => {
        const data = []
        const fields = ['identifier', 'name', 'contact', 'group', 'score']
        _.forEach(criteria, (criterion, idx) => {
          fields.push(`Criterion${idx + 1}`)
        })
        _.forEach(students, student => {
          const record = {
            identifier: student.identifier,
            name: student.name,
            contact: student.contact,
            score: student.score
          }
          if (_.has(student, 'group_id')) {
            _.forEach(groups, (group, idx) => {
              if (student.group_id === group._id) {
                record['group'] = `Group${idx + 1}`
                return
              }
            })
          }
          // let counter = 1;
          _.forEach(criteria, (criterion, idx) => {
            const header = `Criterion${idx + 1}`
            record[header] = _.includes(student.criterion_ids, criterion._id) ? 1 : 0
          })
          data.push(record)
        })
        // console.log(data);

        return convertObjectToCsv({data, fields})
      })
    }

    /*
        http://stackoverflow.com/questions/16245767/creating-a-blob-from-a-base64-string-in-javascript
        */
    let base64toBlob = (b64Data, contentType, sliceSize) => {
      contentType = contentType || ''
      sliceSize = sliceSize || 512

      var byteCharacters = atob(b64Data)
      var byteArrays = []

      for (var offset = 0; offset < byteCharacters.length; offset += sliceSize) {
        var slice = byteCharacters.slice(offset, offset + sliceSize)

        var byteNumbers = new Array(slice.length)
        for (var i = 0; i < slice.length; i++) {
          byteNumbers[i] = slice.charCodeAt(i)
        }

        var byteArray = new Uint8Array(byteNumbers)

        byteArrays.push(byteArray)
      }

      var blob = new Blob(byteArrays, {type: contentType})
      return blob
    }

    this.creatGradingSheetBlob = () => {
      let promises = [ DataService.findAllCriteria(), DataService.findAllStudents() ]
      return $q.all(promises).then(( [criteria, students] ) => {
        // if (students.length == 0) {
        //     return $q.reject(new Error("there is no student record in order to export"));
        // }
        const data = []
        _.forEach(students, student => {
          let feedback = []
          _.forEach(criteria, criterion => {
            if (_.includes(student.criterion_ids, criterion._id)) {

              const safeHTML = (`(${criterion.points}) : ${criterion.description}`)
                .replace(/&/g, '&')
                .replace(/</g, '<')
                .replace(/>/g, '>')
                .replace(/"/g, '"')

              feedback.push(`<p>${safeHTML}</p>`)
              // feedback.push(`(${criterion.points}) : ${criterion.description}`);
            }
          })

          const isNumeric = e => !isNaN(parseFloat(e)) && isFinite(e)
          const exportedTime = moment().format('dddd, MMMM DD, YYYY, h:mm A')

          data.push({
            identifier: `${student.identifier}`,
            name: `${student.name}`,
            grade: isNumeric(student.score) ? student.score : 0,
            exported_time: exportedTime,
            feedback: `${feedback.join('')} `
          })
        })

        const fields = ['identifier', 'name', 'grade', 'exported_time', 'feedback']
        return convertObjectToCsv({ data, fields }).then(csv => {
          return new Blob([csv], { type: 'text/csv' })
        })

        // const fields = ['identifier', 'name', 'grade', 'exported_time', 'feedback'];
        // const fieldNames = ['identifier', 'name', 'grade', 'exported_time', 'feedback']; // TODO: redundant
        // json2csv({ data, fields, fieldNames }, (err, csv) => {
        //     if (err) {
        //         console.log(err);
        //         $mdToast.showSimple('Error on generating CSV file');
        //     } else {
        //         $mdToast.showSimple('CSV file is generated');
        //         SaveFileAs(`${this.databaseName}.csv`, 'text/csv', csv);
        //     }
        // });

      })
    }
  }

})()


text:
using System;
using System.Collections.ObjectModel;
using System.Diagnostics;
using System.IO;
using System.Management.Automation;
using System.Management.Automation.Internal;
using System.Management.Automation.Runspaces;
using System.Threading;

namespace Microsoft.PowerShell.Commands
{
    /// <summary>
    /// A cmdlet that traces the specified categories and flags for the duration of the
    /// specified expression.
    /// </summary>
    [Cmdlet(VerbsDiagnostic.Trace, "Command", DefaultParameterSetName = "expressionSet", HelpUri = "https://go.microsoft.com/fwlink/?LinkID=2097136")]
    public class TraceCommandCommand : TraceListenerCommandBase, IDisposable
    {
        #region Parameters

        /// <summary>
        /// This parameter specifies the current pipeline object.
        /// </summary>
        [Parameter(ValueFromPipeline = true)]
        public PSObject InputObject { get; set; } = AutomationNull.Value;

        /// <summary>
        /// The TraceSource parameter determines which TraceSource categories the
        /// operation will take place on.
        /// </summary>
        [Parameter(Position = 0, Mandatory = true)]
        public string[] Name
        {
            get { return base.NameInternal; }

            set { base.NameInternal = value; }
        }

        /// <summary>
        /// The flags to be set on the TraceSource.
        /// </summary>
        /// <value></value>
        [Parameter(Position = 2)]
        public PSTraceSourceOptions Option
        {
            get
            {
                return base.OptionsInternal;
            }

            set
            {
                base.OptionsInternal = value;
            }
        }

        /// <summary>
        /// The parameter for the expression that should be traced.
        /// </summary>
        /// <value></value>
        [Parameter(Position = 1, Mandatory = true, ParameterSetName = "expressionSet")]
        public ScriptBlock Expression { get; set; }

        /// <summary>
        /// The parameter for the expression that should be traced.
        /// </summary>
        /// <value></value>
        [Parameter(Position = 1, Mandatory = true, ParameterSetName = "commandSet")]
        public string Command { get; set; }

        /// <summary>
        /// When set, this parameter is the arguments to pass to the command specified by
        /// the -Command parameter.
        /// </summary>
        [Parameter(ParameterSetName = "commandSet", ValueFromRemainingArguments = true)]
        [Alias("Args")]
        public object[] ArgumentList { get; set; }

        /// <summary>
        /// The parameter which determines the options for output from the trace listeners.
        /// </summary>
        [Parameter]
        public TraceOptions ListenerOption
        {
            get
            {
                return base.ListenerOptionsInternal;
            }

            set
            {
                base.ListenerOptionsInternal = value;
            }
        }

        /// <summary>
        /// Adds the file trace listener using the specified file.
        /// </summary>
        /// <value></value>
        [Parameter]
        [Alias("PSPath", "Path")]
        public string FilePath
        {
            get { return base.FileListener; }

            set { base.FileListener = value; }
        }

        /// <summary>
        /// Force parameter to control read-only files.
        /// </summary>
        [Parameter]
        public SwitchParameter Force
        {
            get { return base.ForceWrite; }

            set { base.ForceWrite = value; }
        }

        /// <summary>
        /// If this parameter is specified the Debugger trace listener will be added.
        /// </summary>
        /// <value></value>
        [Parameter]
        public SwitchParameter Debugger
        {
            get { return base.DebuggerListener; }

            set { base.DebuggerListener = value; }
        }

        /// <summary>
        /// If this parameter is specified the Msh Host trace listener will be added.
        /// </summary>
        /// <value></value>
        [Parameter]
        public SwitchParameter PSHost
        {
            get { return base.PSHostListener; }

            set { base.PSHostListener = value; }
        }

        #endregion Parameters

        #region Cmdlet code

        private Collection<PSTraceSource> _matchingSources;

        /// <summary>
        /// Gets the PSTraceSource instances that match the names specified.
        /// </summary>
        protected override void BeginProcessing()
        {
            Collection<PSTraceSource> preconfiguredSources = null;
            _matchingSources = ConfigureTraceSource(base.NameInternal, false, out preconfiguredSources);

            TurnOnTracing(_matchingSources, false);
            TurnOnTracing(preconfiguredSources, true);

            // Now that tracing has been configured, move all the sources into a
            // single collection

            foreach (PSTraceSource preconfiguredSource in preconfiguredSources)
            {
                _matchingSources.Add(preconfiguredSource);
            }

            if (ParameterSetName == "commandSet")
            {
                // Create the CommandProcessor and add it to a pipeline

                CommandProcessorBase commandProcessor =
                    this.Context.CommandDiscovery.LookupCommandProcessor(Command, CommandOrigin.Runspace, false);

                // Add the parameters that were specified

                ParameterBinderController.AddArgumentsToCommandProcessor(commandProcessor, ArgumentList);

                _pipeline = new PipelineProcessor();
                _pipeline.Add(commandProcessor);

                // Hook up the success and error pipelines to this cmdlet's WriteObject and
                // WriteError methods

                _pipeline.ExternalErrorOutput = new TracePipelineWriter(this, true, _matchingSources);
                _pipeline.ExternalSuccessOutput = new TracePipelineWriter(this, false, _matchingSources);
            }

            ResetTracing(_matchingSources);
        }

        /// <summary>
        /// Executes the expression.
        /// Note, this was taken from apply-expression.
        /// </summary>
        protected override void ProcessRecord()
        {
            TurnOnTracing(_matchingSources, false);

            object result = null;
            switch (ParameterSetName)
            {
                case "expressionSet":
                    result = RunExpression();
                    break;

                case "commandSet":
                    result = StepCommand();
                    break;
            }

            ResetTracing(_matchingSources);

            if (result == null)
            {
                return;
            }

            if (!LanguagePrimitives.IsNull(result))
            {
                WriteObject(result, true);
            }
        }

        /// <summary>
        /// Finishes running the command if specified and then sets the
        /// tracing options and listeners back to their original values.
        /// </summary>
        protected override void EndProcessing()
        {
            if (_pipeline != null)
            {
                TurnOnTracing(_matchingSources, false);

                Array results = _pipeline.SynchronousExecuteEnumerate(AutomationNull.Value);

                ResetTracing(_matchingSources);

                WriteObject(results, true);
            }

            this.Dispose();
        }

        /// <summary>
        /// Ensures that the sub-pipeline we created gets stopped as well.
        /// </summary>
        protected override void StopProcessing()
        {
            if (_pipeline != null)
            {
                _pipeline.Stop();
            }
        }

        #endregion Cmdlet code

        private object RunExpression()
        {
            return Expression.DoInvokeReturnAsIs(
                useLocalScope: false,
                errorHandlingBehavior: ScriptBlock.ErrorHandlingBehavior.WriteToCurrentErrorPipe,
                dollarUnder: InputObject,
                input: new object[] { InputObject },
                scriptThis: AutomationNull.Value,
                args: Array.Empty<object>());
        }

        private object StepCommand()
        {
            if (InputObject != AutomationNull.Value)
            {
                _pipeline.Step(InputObject);
            }

            return null;
        }

        private PipelineProcessor _pipeline;

        #region IDisposable

        /// <summary>
        /// Resets the TraceSource flags back to their original value and restores
        /// the original TraceListeners.
        /// </summary>
        public void Dispose()
        {
            if (!_disposed)
            {
                _disposed = true;

                // Reset the flags for the trace switch back to the original value
                ResetTracing(_matchingSources);
                ClearStoredState();
                _matchingSources = null;

                if (_pipeline != null)
                {
                    _pipeline.Dispose();
                    _pipeline = null;
                }

                // If there are any file streams, close those as well.

                if (this.FileStreams != null)
                {
                    foreach (FileStream fileStream in this.FileStreams)
                    {
                        fileStream.Flush();
                        fileStream.Dispose();
                    }
                }

                GC.SuppressFinalize(this);
            }
        }

        private bool _disposed;
        #endregion IDisposable
    }

    /// <summary>
    /// This class acts a pipe redirector for the sub-pipeline created by the Trace-Command
    /// cmdlet.  It gets attached to the sub-pipelines success or error pipeline and redirects
    /// all objects written to these pipelines to trace-command pipeline.
    /// </summary>
    internal class TracePipelineWriter : PipelineWriter
    {
        internal TracePipelineWriter(
            TraceListenerCommandBase cmdlet,
            bool writeError,
            Collection<PSTraceSource> matchingSources)
        {
            if (cmdlet == null)
            {
                throw new ArgumentNullException(nameof(cmdlet));
            }

            if (matchingSources == null)
            {
                throw new ArgumentNullException(nameof(matchingSources));
            }

            _cmdlet = cmdlet;
            _writeError = writeError;
            _matchingSources = matchingSources;
        }

        /// <summary>
        /// Get the wait handle signaled when buffer space is available in the underlying stream.
        /// </summary>
        public override WaitHandle WaitHandle
        {
            get { return null; }
        }

        /// <summary>
        /// Check if the stream is open for further writes.
        /// </summary>
        /// <value>true if the underlying stream is open, otherwise; false.</value>
        /// <remarks>
        /// Attempting to write to the underlying stream if IsOpen is false throws
        /// an <see cref="ObjectDisposedException"/>.
        /// </remarks>
        public override bool IsOpen
        {
            get
            {
                return _isOpen;
            }
        }

        /// <summary>
        /// Returns the number of objects in the underlying stream.
        /// </summary>
        public override int Count
        {
            get { return 0; }
        }

        /// <summary>
        /// Get the capacity of the stream.
        /// </summary>
        /// <value>
        /// The capacity of the stream.
        /// </value>
        /// <remarks>
        /// The capacity is the number of objects that stream may contain at one time.  Once this
        /// limit is reached, attempts to write into the stream block until buffer space
        /// becomes available.
        /// </remarks>
        public override int MaxCapacity
        {
            get { return int.MaxValue; }
        }

        /// <summary>
        /// Close the stream.
        /// </summary>
        /// <remarks>
        /// Causes subsequent calls to IsOpen to return false and calls to
        /// a write operation to throw an ObjectDisposedException.
        /// All calls to Close() after the first call are silently ignored.
        /// </remarks>
        /// <exception cref="ObjectDisposedException">
        /// The stream is already disposed.
        /// </exception>
        public override void Close()
        {
            if (_isOpen)
            {
                Flush();
                _isOpen = false;
            }
        }

        /// <summary>
        /// Flush the data from the stream.  Closed streams may be flushed,
        /// but disposed streams may not.
        /// </summary>
        /// <exception cref="ObjectDisposedException">
        /// The underlying stream is disposed.
        /// </exception>
        public override void Flush()
        {
        }

        /// <summary>
        /// Write a single object into the underlying stream.
        /// </summary>
        /// <param name="obj">The object to add to the stream.</param>
        /// <returns>
        /// One, if the write was successful, otherwise;
        /// zero if the stream was closed before the object could be written,
        /// or if the object was AutomationNull.Value.
        /// </returns>
        /// <exception cref="ObjectDisposedException">
        /// The underlying stream is closed.
        /// </exception>
        public override int Write(object obj)
        {
            _cmdlet.ResetTracing(_matchingSources);

            if (_writeError)
            {
                ErrorRecord errorRecord = ConvertToErrorRecord(obj);

                if (errorRecord != null)
                {
                    _cmdlet.WriteError(errorRecord);
                }
            }
            else
            {
                _cmdlet.WriteObject(obj);
            }

            _cmdlet.TurnOnTracing(_matchingSources, false);
            return 1;
        }

        /// <summary>
        /// Write objects to the underlying stream.
        /// </summary>
        /// <param name="obj">Object or enumeration to read from.</param>
        /// <param name="enumerateCollection">
        /// If enumerateCollection is true, and <paramref name="obj"/>
        /// is an enumeration according to LanguagePrimitives.GetEnumerable,
        /// the objects in the enumeration will be unrolled and
        /// written separately.  Otherwise, <paramref name="obj"/>
        /// will be written as a single object.
        /// </param>
        /// <returns>The number of objects written.</returns>
        /// <exception cref="ObjectDisposedException">
        /// The underlying stream is closed.
        /// </exception>
        /// <exception cref="ArgumentException">
        /// <paramref name="obj"/> contains AutomationNull.Value
        /// </exception>
        public override int Write(object obj, bool enumerateCollection)
        {
            _cmdlet.ResetTracing(_matchingSources);

            int numWritten = 0;
            if (_writeError)
            {
                if (enumerateCollection)
                {
                    foreach (object o in LanguagePrimitives.GetEnumerable(obj))
                    {
                        ErrorRecord errorRecord = ConvertToErrorRecord(o);
                        if (errorRecord != null)
                        {
                            numWritten++;
                            _cmdlet.WriteError(errorRecord);
                        }
                    }
                }
                else
                {
                    ErrorRecord errorRecord = ConvertToErrorRecord(obj);
                    if (errorRecord != null)
                    {
                        numWritten++;
                        _cmdlet.WriteError(errorRecord);
                    }
                }
            }
            else
            {
                numWritten++;
                _cmdlet.WriteObject(obj, enumerateCollection);
            }

            _cmdlet.TurnOnTracing(_matchingSources, false);

            return numWritten;
        }

        private static ErrorRecord ConvertToErrorRecord(object obj)
        {
            ErrorRecord result = null;
            PSObject mshobj = obj as PSObject;
            if (mshobj != null)
            {
                object baseObject = mshobj.BaseObject;
                if (baseObject is not PSCustomObject)
                {
                    obj = baseObject;
                }
            }

            ErrorRecord errorRecordResult = obj as ErrorRecord;
            if (errorRecordResult != null)
            {
                result = errorRecordResult;
            }

            return result;
        }

        private readonly TraceListenerCommandBase _cmdlet;
        private readonly bool _writeError;
        private bool _isOpen = true;
        private readonly Collection<PSTraceSource> _matchingSources = new();
    }
}


text:
using System;
using System.Linq;
using System.Threading.Tasks;
using Marten.Testing.Harness;
using Shouldly;
using Xunit;

namespace EventSourcingTests.Bugs;

public class Bug_2011_streams_should_be_getting_correct_timestamp : BugIntegrationContext
{
    [Fact]
    public async Task capture_multiple_streams_and_check_created()
    {
        var stream1 = Guid.NewGuid();
        var stream2 = Guid.NewGuid();
        var stream3 = Guid.NewGuid();
        var stream4 = Guid.NewGuid();

        theSession.Events.Append(stream1, new QuestStarted { Name = "One" });
        await theSession.SaveChangesAsync();

        await Task.Delay(25);

        theSession.Events.Append(stream2, new QuestStarted { Name = "Two" });
        await theSession.SaveChangesAsync();

        await Task.Delay(500);

        theSession.Events.Append(stream3, new QuestStarted { Name = "Three" });
        await theSession.SaveChangesAsync();

        await Task.Delay(500);

        theSession.Events.Append(stream4, new QuestStarted { Name = "Four" });
        await theSession.SaveChangesAsync();

        var s1 = await theSession.Events.FetchStreamStateAsync(stream1);
        var s2 = await theSession.Events.FetchStreamStateAsync(stream2);
        var s3 = await theSession.Events.FetchStreamStateAsync(stream3);
        var s4 = await theSession.Events.FetchStreamStateAsync(stream4);

        var dates = new DateTimeOffset[] { s1.Created, s2.Created, s3.Created, s4.Created };
        dates.Distinct().Count().ShouldBe(4);
    }

}


text:
#include "TableModelForwardLook.h"

// Qt
#include <QLabel>
#include <QAbstractItemView>

// Custom
#include "NamedITKImageCollection.h"
#include "Helpers/Helpers.h"
#include "HelpersQt.h"

TableModelForwardLook::TableModelForwardLook(QObject * parent, std::vector<InpaintingIterationRecord> const& iterationRecords, DisplayStyle const& style) :
    QAbstractTableModel(parent), IterationRecords(iterationRecords), ImageDisplayStyle(style), IterationToDisplay(0), PatchDisplaySize(100)
{
}

void TableModelForwardLook::SetPatchDisplaySize(const unsigned int value)
{
  this->PatchDisplaySize = value;
}

unsigned int TableModelForwardLook::GetPatchDisplaySize()
{
  return this->PatchDisplaySize;
}

Qt::ItemFlags TableModelForwardLook::flags(const QModelIndex& index) const
{
  //Qt::ItemFlags itemFlags = (!Qt::ItemIsEditable) | Qt::ItemIsSelectable | Qt::ItemIsEnabled | (!Qt::ItemIsUserCheckable) | (!Qt::ItemIsTristate);
  Qt::ItemFlags itemFlags = Qt::ItemIsSelectable | Qt::ItemIsEnabled;
  return itemFlags;
}

void TableModelForwardLook::SetIterationToDisplay(const unsigned int iteration)
{
  this->IterationToDisplay = iteration;
  Refresh();
}

int TableModelForwardLook::rowCount(const QModelIndex& parent) const
{
  EnterFunction("ForwardLookTableModel::rowCount()");
  if(this->IterationRecords.size() < this->IterationToDisplay ||
    this->IterationRecords.size() == 0)
    {
    return 0;
    }
  unsigned int rows = this->IterationRecords[this->IterationToDisplay].PotentialPairSets.size();
  LeaveFunction("ForwardLookTableModel::rowCount()");
  return rows;
}

int TableModelForwardLook::columnCount(const QModelIndex& parent) const
{
  return 3;
}

QVariant TableModelForwardLook::data(const QModelIndex& index, int role) const
{
  EnterFunction("TableModelForwardLook::data()");
  QVariant returnValue;
  if(role == Qt::DisplayRole && index.row() >= 0)
    {
    const CandidatePairs& currentCandidateSet = *(this->IterationRecords[this->IterationToDisplay].PotentialPairSets[index.row()]);
    const Patch currentForwardLookPatch = currentCandidateSet.GetTargetPatch();
    switch(index.column())
      {
      case 0:
        {
        // Display the target patch in the table
        InpaintingIterationRecord currentRecord = this->IterationRecords[this->IterationToDisplay];
        FloatVectorImageType* image = dynamic_cast<FloatVectorImageType*>(currentRecord.GetImageByName("Image").Image);
        QImage patchImage = HelpersQt::GetQImage<FloatVectorImageType>(image, currentForwardLookPatch.GetRegion(), this->ImageDisplayStyle);

        patchImage = patchImage.scaledToHeight(this->PatchDisplaySize);

        returnValue = QPixmap::fromImage(patchImage);
        break;
        }
      case 1:
        {
        // Display priority in the table
        returnValue = currentCandidateSet.GetPriority();
        break;
        }
      case 2:
        {
        // Display the target patch location in the table
        std::stringstream ssLocation;
        ssLocation << "(" << currentForwardLookPatch.GetRegion().GetIndex()[0] << ", " << currentForwardLookPatch.GetRegion().GetIndex()[1] << ")";
        returnValue = ssLocation.str().c_str();
        break;
        }
      } // end switch
    } // end if DisplayRole
  LeaveFunction("ForwardLookTableModel::data()");
  return returnValue;
}

QVariant TableModelForwardLook::headerData(int section, Qt::Orientation orientation, int role) const
{
  QVariant returnValue;
  if(role == Qt::DisplayRole)
    {
    if(orientation == Qt::Horizontal)
      {
      switch(section)
	{
	case 0:
	  returnValue = "Patch";
	  break;
	case 1:
	  returnValue = "Priority";
	  break;
	case 2:
	  returnValue = "Location";
	  break;
	} // end switch
      }// end Horizontal orientation
    } // end DisplayRole

  return returnValue;
}

void TableModelForwardLook::Refresh()
{
  //std::cout << "ForwardLookTableModel::Refresh(): Displaying iteration: " << this->IterationToDisplay << std::endl;
  beginResetModel();
  endResetModel();
}

void TableModelForwardLook::selectionChanged(const QItemSelection& selected, const QItemSelection& deselected)
{
  //std::cout << "ForwardLookTableModel::selectionChanged()" << std::endl;
}


text:
<?php

// autoload.php @generated by Composer

require_once __DIR__ . '/composer' . '/autoload_real.php';

return ComposerAutoloaderInitb61412e7fb1100e5178a5df7659cdc41::getLoader();


text:
#region Copyright (c) 2002-2005, Bas Geertsema, Xih Solutions (http://www.xihsolutions.net)

#endregion

using System;
using System.IO;
using System.Collections;
using System.Text;
using System.Globalization;
using System.Threading;
using System.Net;
using XihSolutions.DotMSN.Core;
using XihSolutions.DotMSN;

namespace XihSolutions.DotMSN.DataTransfer
{
	/// <summary>
	/// P2PMessageSession routes all messages in the p2p framework between the local client and a single remote client.
	/// </summary>
	/// <remarks>
	/// A single message session can hold multiple p2p transfer sessions. This for example occurs when a contact sends
	/// two files directly after each other in the same switchboard session.
	/// This class keeps track of the message identifiers, dispatches messages to registered message handlers and routes
	/// data messages to the correct <see cref="P2PTransferSession"/> objects. Usually this class is a handler of a switchboard processor.
	/// A common handler for this class is <see cref="MSNSLPHandler"/>.
	/// </remarks>
	public class P2PMessageSession : IMessageHandler, IMessageProcessor
	{
		#region Properties
		/// <summary>
		/// </summary>
		private uint localBaseIdentifier;
		/// <summary>
		/// </summary>
		private uint localIdentifier;
		/// <summary>
		/// </summary>
		private uint remoteBaseIdentifier;
		/// <summary>
		/// </summary>
		private uint remoteIdentifier;

		/// <summary>
		/// </summary>
		private P2PDCHandshakeMessage handshakeMessage;

		/// <summary>
		/// The handshake message to send to the receiving client when a direct connection has been established
		/// </summary>
		/// <remarks>
		/// If this property is set to null no handshake message is sent.
		/// </remarks>
		public P2PDCHandshakeMessage HandshakeMessage
		{
			get { return handshakeMessage; }
			set { handshakeMessage = value;}
		}

		/// <summary>
		/// </summary>
		private bool autoAcknowledgeHandshake = true;

		/// <summary>
		/// Defines whether a direct connection handshake is automatically send to the remote client, or replied with an acknowledgement.
		/// Setting this to true means the remote client will start the transfer immediately.
		/// Setting this to false means the client programmer must send a handhsake message and an acknowledgement message after which the transfer will begin.
		/// </summary>
		public bool AutoHandshake
		{
			get { return autoAcknowledgeHandshake; }
			set { autoAcknowledgeHandshake = value;}
		}

		/// <summary>
		/// </summary>
		private bool directConnected = false;

		/// <summary>
		/// Defines whether the message session runs over a direct session or is routed via the messaging server
		/// </summary>
		public bool DirectConnected
		{
			get { return directConnected; }
		}

		/// <summary>
		/// </summary>
		private bool directConnectionAttempt = false;

		/// <summary>
		/// Defines whether an attempt has been made to create a direct connection
		/// </summary>
		public bool DirectConnectionAttempt
		{
			get { return directConnectionAttempt; }
		}

		/// <summary>
		/// Occurs when a direct connection is succesfully established.
		/// </summary>
		public event EventHandler DirectConnectionEstablished;

		/// <summary>
		/// Occurs when a direct connection attempt has failed.
		/// </summary>
		public event EventHandler DirectConnectionFailed;

		/// <summary>
		/// The base identifier of the local client
		/// </summary>
		public uint LocalBaseIdentifier
		{
			get { return localBaseIdentifier; }
			set { localBaseIdentifier = value;}
		}

		/// <summary>
		/// The identifier of the local contact. This identifier is increased just before a message is send.
		/// </summary>
		public uint LocalIdentifier
		{
			get { return localIdentifier; }
			set { localIdentifier = value;}
		}

		/// <summary>
		/// The base identifier of the remote client
		/// </summary>
		public uint RemoteBaseIdentifier
		{
			get { return remoteBaseIdentifier; }
			set { remoteBaseIdentifier = value;}
		}
		/// <summary>
		/// The expected identifier of the remote client for the next message.
		/// </summary>
		public uint RemoteIdentifier
		{
			get { return remoteIdentifier; }
			set { remoteIdentifier = value;}
		}

		/// <summary>
		/// </summary>
		private string remoteContact;
		/// <summary>
		/// </summary>
		private string localContact;

		/// <summary>
		/// The account of the local contact.
		/// </summary>
		public string LocalContact
		{
			get { return localContact; }
			set { localContact = value;}
		}

		/// <summary>
		/// The account of the remote contact.
		/// </summary>
		public string RemoteContact
		{
			get { return remoteContact;}
			set { remoteContact = value;}
		}

		#endregion

		#region Public
		/// <summary>
		/// Constructor.
		/// </summary>
		public P2PMessageSession()
		{
			if(Settings.TraceSwitch.TraceInfo)
				System.Diagnostics.Trace.WriteLine("Constructing object", "P2PMessageSession");
		}

		/// <summary>
		/// Removes references to handlers and the messageprocessor. Also closes running transfer sessions and pending processors establishing connections.
		/// </summary>
		public virtual void CleanUp()
		{
            StopAllPendingProcessors();
            AbortAllTransfers();

			handlers.Clear();

            MessageProcessor = null;

			transferSessions.Clear();
		}

        /// <summary>
        /// A list of all direct processors trying to establish a connection.
        /// </summary>
        private ArrayList pendingProcessors = new ArrayList();

        /// <summary>
        /// Disconnect all processors that are trying to establish a connection.
        /// </summary>
        protected void StopAllPendingProcessors()
        {
            lock (pendingProcessors)
            {
                foreach (P2PDirectProcessor processor in pendingProcessors)
                {
                    processor.Disconnect();
                    processor.UnregisterHandler(this);
                }
                pendingProcessors.Clear();
            }
        }

        /// <summary>
        /// Add the processor to the pending list.
        /// </summary>
        /// <param name="processor"></param>
        protected void AddPendingProcessor(P2PDirectProcessor processor)
        {
            // we want to handle message from this processor
            processor.RegisterHandler(this);

            // inform the session of connected/disconnected events
            processor.ConnectionEstablished += new EventHandler(OnDirectProcessorConnected);
            processor.ConnectionClosed += new EventHandler(OnDirectProcessorDisconnected);
            processor.ConnectingException += new ProcessorExceptionEventHandler(OnDirectProcessorException);

            lock (pendingProcessors)
            {
                pendingProcessors.Add(processor);
            }
        }

        /// <summary>
        /// Use the given processor as the direct connection processor. And disconnect all other pending processors.
        /// </summary>
        /// <param name="processor"></param>
        protected void UsePendingProcessor(P2PDirectProcessor processor)
        {
            lock (pendingProcessors)
            {
                if (pendingProcessors.Contains(processor))
                {
                    pendingProcessors.Remove(processor);
                }
            }

            // stop all remaining attempts
            StopAllPendingProcessors();

            // set the direct processor as the main processor
            lock (this)
            {
                directConnected = true;
                directConnectionAttempt = true;
                preDCProcessor = MessageProcessor;
                MessageProcessor = processor;
            }

            if (DirectConnectionEstablished != null)
                DirectConnectionEstablished(this, new EventArgs());
        }

		/// <summary>
		/// Creates a direct connection with the remote client.
		/// </summary>
		/// <param name="host"></param>
		/// <param name="port"></param>
		/// <returns></returns>
		public IMessageProcessor CreateDirectConnection(string host, int port)
		{
			// create the P2P Direct processor to handle the file data
			P2PDirectProcessor processor = new P2PDirectProcessor();
			processor.ConnectivitySettings = new ConnectivitySettings();
			processor.ConnectivitySettings.Host = host;
			processor.ConnectivitySettings.Port = port;

			if(Settings.TraceSwitch.TraceInfo)
				System.Diagnostics.Trace.WriteLine("Trying to setup direct connection with remote host " + host + ":" + port.ToString(System.Globalization.CultureInfo.InvariantCulture), "P2PTransferSession");

            AddPendingProcessor(processor);

            // try to connect
			processor.Connect();

			return processor;
		}

		/// <summary>
		/// Setups a P2PDirectProcessor to listen for incoming connections.
		/// After a connection has been established the P2PDirectProcessor will become the main MessageProcessor to send messages.
		/// </summary>
		/// <param name="host"></param>
		/// <param name="port"></param>
		/// <returns></returns>
		public IMessageProcessor ListenForDirectConnection(IPAddress host, int port)
		{
			P2PDirectProcessor processor = new P2PDirectProcessor();

			// we want this session to listen to incoming messages
			processor.RegisterHandler(this);

            // add to the list of processors trying to establish a connection
            AddPendingProcessor(processor);

            // start to listen
            processor.Listen(host, port);

			return processor;
		}

		/// <summary>
		/// Closes the direct connection with the remote client, if available. A closing p2p message will be send first.
		/// The session will fallback to the previous (SB) message processor.
		/// </summary>
		public void CloseDirectConnection()
		{
			if(DirectConnected == false)
				return;
			else
			{
				CleanUpDirectConnection();
			}
		}

		/// <summary>
		/// Aborts all running transfer sessions.
		/// </summary>
		public virtual void AbortAllTransfers()
		{
			foreach(P2PTransferSession session in transferSessions.Values)
			{
				session.AbortTransfer();
			}
		}

		/// <summary>
		/// Corrects the local identifier with the specified correction.
		/// </summary>
		/// <param name="correction"></param>
		public void CorrectLocalIdentifier(int correction)
		{
			if(correction < 0)
				LocalIdentifier -= (uint)Math.Abs(correction);
			else
				LocalIdentifier += (uint)Math.Abs(correction);
		}

		/// <summary>
		/// The identifier of the local client, increases with each message send
		/// </summary>
		public void IncreaseLocalIdentifier()
		{
			localIdentifier++;
			if(localIdentifier == localBaseIdentifier)
				localIdentifier++;
		}

		/// <summary>
		/// The identifier of the remote client, increases with each message received
		/// </summary>
		public void IncreaseRemoteIdentifier()
		{
			remoteIdentifier++;
			if(remoteIdentifier == remoteBaseIdentifier)
				remoteIdentifier++;
		}

		/// <summary>
		/// Adds the specified transfer session to the collection and sets the transfer session's message processor to be the
		/// message processor of the p2p message session. This is usally a SB message processor.
		/// </summary>
		/// <param name="session"></param>
		public void AddTransferSession(P2PTransferSession session)
		{
			session.MessageProcessor = this;

			lock(handlers)
			{
				foreach(IMessageHandler handler in handlers)
				{
					session.RegisterHandler(handler);
				}
			}

			transferSessions.Add(session.SessionId, session);
		}

		/// <summary>
		/// Removes the specified transfer session from the collection.
		/// </summary>
		public void RemoveTransferSession(P2PTransferSession session)
		{
			transferSessions.Remove(session);
		}

		/// <summary>
		/// Returns the transfer session associated with the specified session identifier.
		/// </summary>
		/// <param name="sessionId"></param>
		/// <returns></returns>
		public P2PTransferSession GetTransferSession(uint sessionId)
		{
			return (P2PTransferSession)transferSessions[sessionId];
		}

		/// <summary>
		/// Searches through all handlers and returns the first object with the specified type, or null if not found.
		/// </summary>
		/// <param name="handlerType"></param>
		/// <returns></returns>
		public object GetHandler(Type handlerType)
		{
			foreach(IMessageHandler handler in handlers)
			{
				if(handler.GetType() == handlerType)
					return handler;
			}
			return null;
		}

		#endregion

		#region Protected

		/// <summary>
		/// Keeps track of clustered p2p messages
		/// </summary>
		protected P2PMessagePool P2PMessagePool
		{
			get { return p2pMessagePool; }
			set { p2pMessagePool = value;}
		}

		/// <summary>
		/// </summary>
		private P2PMessagePool p2pMessagePool = new P2PMessagePool();

		/// <summary>
		/// Wraps a P2PMessage in a MSGMessage and SBMessage.
		/// </summary>
		/// <returns></returns>
		protected SBMessage WrapMessage(NetworkMessage networkMessage)
		{
			// create wrapper messages
			MSGMessage msgWrapper = new MSGMessage();
			msgWrapper.MimeHeader["P2P-Dest"] = RemoteContact;
			msgWrapper.MimeHeader["Content-Type"] = "application/x-msnmsgrp2p";
			msgWrapper.InnerMessage = networkMessage;

			SBMessage sbMessageWrapper = new SBMessage();
			sbMessageWrapper.InnerMessage = msgWrapper;

			return sbMessageWrapper;
		}

		/// <summary>
		/// Sends the handshake message in a direct connection.
		/// </summary>
		protected virtual void SendHandshakeMessage(IMessageProcessor processor)
		{
            if (Settings.TraceSwitch.TraceError)
                System.Diagnostics.Trace.WriteLine("Preparing to send handshake message", "P2PMessageSession");

            if(HandshakeMessage == null)
			{
				// don't throw an exception because the file transfer can continue over the switchboard
				if(Settings.TraceSwitch.TraceError)
					System.Diagnostics.Trace.WriteLine("Handshake could not be send because none is specified.");

				// but close the direct connection
				((SocketMessageProcessor)processor).Disconnect();
				return;
			}

			IncreaseLocalIdentifier();
			HandshakeMessage.Identifier = LocalIdentifier;

			HandshakeMessage.AckSessionId = (uint)new Random().Next(50000, int.MaxValue);
			DCHandshakeAck = HandshakeMessage.AckSessionId;

			if(Settings.TraceSwitch.TraceInfo)
				System.Diagnostics.Trace.WriteLine("Sending handshake message:\r\n " + HandshakeMessage.ToDebugString(), "P2PMessageSession");

            ((SocketMessageProcessor)processor).SendMessage(HandshakeMessage);
		}

		#endregion

		#region Private
		/// <summary>
		/// A collection of all transfersessions
		/// </summary>
		private Hashtable transferSessions = new Hashtable();

		/// <summary>
		/// This is the processor used before a direct connection. Usually a SB processor.
		/// It is a fallback variables in case a direct connection fails.
		/// </summary>
		private IMessageProcessor	preDCProcessor = null;

		/// <summary>
		/// Tracked to know when an acknowledgement for the handshake is received.
		/// </summary>
		private uint	DCHandshakeAck = 0;

		/// <summary>
		/// Sets the message processor back to the switchboard message processor.
		/// </summary>
		private void CleanUpDirectConnection()
		{
			lock(this)
			{
				if(DirectConnected == false)
					return;

				SocketMessageProcessor directProcessor = (SocketMessageProcessor)MessageProcessor;

				directConnected = false;
				directConnectionAttempt = false;
				MessageProcessor = preDCProcessor;

				directProcessor.Disconnect();
			}
		}

		/// <summary>
		/// Sets the current message processor to the processor which has just connected succesfully.
		/// </summary>
		/// <param name="sender"></param>
		/// <param name="e"></param>
		private void OnDirectProcessorConnected(object sender, EventArgs e)
		{
			//DCHandshakeProcessor = (IMessageProcessor)sender;

            if (((P2PDirectProcessor)sender).IsListener == false)
            {
                if (AutoHandshake == true && HandshakeMessage != null)
                {
                    SendHandshakeMessage((P2PDirectProcessor)sender);
                }
            }
		}

		/// <summary>
		/// Cleans up the direct connection.
		/// </summary>
		/// <param name="sender"></param>
		/// <param name="e"></param>
		private void OnDirectProcessorDisconnected(object sender, EventArgs e)
		{
			CleanUpDirectConnection();
		}

		/// <summary>
		/// Called when the direct processor could not connect. It will start the data transfer over the switchboard session.
		/// </summary>
		/// <param name="sender"></param>
		/// <param name="e"></param>
		private void OnDirectProcessorException(object sender, ExceptionEventArgs e)
		{
			lock(this)
			{
				CleanUpDirectConnection();
				directConnectionAttempt = true;
			}

			if(DirectConnectionFailed != null)
				DirectConnectionFailed(this, new EventArgs());
		}

		/// <summary>
		/// Occurs when an acknowledgement to a send handshake has been received, or a handshake is received.
		/// This will start the data transfer, provided the local client is the sender.
		/// </summary>
		protected virtual void OnHandshakeCompleted(P2PDirectProcessor processor)
		{
			if(Settings.TraceSwitch.TraceInfo)
				System.Diagnostics.Trace.WriteLine("Handshake accepted", "P2PTransferSession");

            UsePendingProcessor(processor);
		}
		#endregion

		#region IMessageHandler Members
		private IMessageProcessor messageProcessor = null;
		/// <summary>
		/// The message processor that sends the P2P messages to the remote contact.
		/// </summary>
		public IMessageProcessor MessageProcessor
		{
			get
			{
				return messageProcessor;
			}
			set
			{
				messageProcessor = value;
				if(messageProcessor != null)
				{
					ValidateProcessor();
					SendBuffer();
				}
			}
		}

		/// <summary>
		/// Handles P2PMessages. Other messages are ignored. All incoming messages are supposed to belong to this session.
		/// </summary>
		public void HandleMessage(IMessageProcessor sender, NetworkMessage message)
		{
			System.Diagnostics.Debug.Assert(message is P2PMessage, "Incoming message is not a P2PMessage", "");

			P2PMessage p2pMessage = (P2PMessage)message;

			// check whether it is an acknowledgement to data preparation message
			if(p2pMessage.Flags == 0x100 && DCHandshakeAck != 0)
			{
				OnHandshakeCompleted((P2PDirectProcessor)sender);
				return;
			}

			// check if it's a direct connection handshake
			if(p2pMessage.Flags == 0x100 && AutoHandshake == true)
			{
				// create a handshake message based on the incoming p2p message and send it
				P2PDCHandshakeMessage dcHsMessage = new P2PDCHandshakeMessage(p2pMessage);
				sender.SendMessage(dcHsMessage.CreateAcknowledgement());
                OnHandshakeCompleted((P2PDirectProcessor)sender);
				return;
			}

			// check if it is a content message
			if(p2pMessage.SessionId > 0)
			{
				// get the session to handle this message
				P2PTransferSession session = (P2PTransferSession)transferSessions[p2pMessage.SessionId];
				if(session != null)
					session.HandleMessage(this, p2pMessage);
				return;
			}

			// it is not a datamessage.
			// fill up the buffer with this message and extract the messages one-by-one and dispatch
			// it to all handlers. Usually the MSNSLP handler.
			p2pMessagePool.BufferMessage(p2pMessage);

			while(p2pMessagePool.MessageAvailable)
			{
				// keep track of the remote identifier
				IncreaseRemoteIdentifier();

				p2pMessage = p2pMessagePool.GetNextMessage();

				lock(handlers)
				{
					// the message is not a datamessage, send it to the handlers
					foreach(IMessageHandler handler in handlers)
						handler.HandleMessage(this, message);
				}
			}
		}

		#endregion

		#region IMessageProcessor Members
		/// <summary>
		/// Sends incoming p2p messages to the remote contact.
		/// </summary>
		/// <remarks>
		/// Before the message is send a couple of things are checked. If there is no identifier available, the local identifier will be increased by one and set as the message identifier.
		/// Second, if the acknowledgement identifier is not set it will be set to a random value. After this the method will check for the total length of the message. If the total length
		/// is too large, the message will be splitted into multiple messages. The maximum size for p2p messages over a switchboard is 1202 bytes. The maximum size for p2p messages over a
		/// direct connection is 1352 bytes. As a result the length of the splitted messages will be 1202 or 1352 bytes or smaller, depending on the availability of a direct connection.
		///
		/// If a direct connection is available the message is wrapped in a <see cref="P2PDCMessage"/> object and send over the direct connection. Otherwise it will be send over a switchboard session.
		/// If there is no switchboard session available, or it has become invalid, a new switchboard session will be requested by asking this to the nameserver handler.
		/// Messages will be buffered until a switchboard session, or a direct connection, becomes available. Upon a new connection the buffered messages are directly send to the remote contact
		/// over the new connection.
		/// </remarks>
		/// <param name="message">The P2PMessage to send to the remote contact.</param>
		public void SendMessage(NetworkMessage message)
		{
			P2PMessage p2pMessage = (P2PMessage)message;

			// check whether it's already set. This is important to check for acknowledge messages.
			if(p2pMessage.Identifier == 0)
			{
				IncreaseLocalIdentifier();
				p2pMessage.Identifier = LocalIdentifier;
			}
			if(p2pMessage.AckSessionId == 0)
			{
				p2pMessage.AckSessionId = (uint)new Random().Next(50000, int.MaxValue);
			}

			// maximum length by default is 1202 bytes.
			int maxSize = 1202;

			// check whether we have a direct connection (send p2pdc messages) or not (send sb messages)
			if(DirectConnected == true)
			{
				maxSize = 1352;
			}

			// split up large messages which go to the SB
			if(p2pMessage.MessageSize > maxSize)
			{
				byte[] totalMessage = null;
				if(p2pMessage.InnerBody != null)
					totalMessage = p2pMessage.InnerBody;
				else if(p2pMessage.InnerMessage != null)
					totalMessage = p2pMessage.InnerMessage.GetBytes();
				else
					throw new DotMSNException("An error occured while splitting a large p2p message into smaller messages: Both the InnerBody and InnerMessage are null.");

				uint bytesSend = 0;
				int cnt = ((int)(p2pMessage.MessageSize / maxSize)) + 1;
				for(int i = 0; i < cnt; i++)
				{
					P2PMessage chunkMessage = new P2PMessage();

					// copy the values from the original message
					chunkMessage.AckIdentifier = p2pMessage.AckIdentifier;
					chunkMessage.AckTotalSize  = p2pMessage.AckTotalSize;
					chunkMessage.Flags         = p2pMessage.Flags;
					chunkMessage.Footer        = p2pMessage.Footer;
					chunkMessage.Identifier    = p2pMessage.Identifier;
					chunkMessage.MessageSize   = (uint)Math.Min((uint)maxSize, (uint)(p2pMessage.MessageSize - bytesSend));
					chunkMessage.Offset        = bytesSend;
					chunkMessage.SessionId     = p2pMessage.SessionId;
					chunkMessage.TotalSize     = p2pMessage.MessageSize;

					chunkMessage.InnerBody = new byte[chunkMessage.MessageSize];
					Array.Copy(totalMessage, (int)chunkMessage.Offset, chunkMessage.InnerBody, 0, (int)chunkMessage.MessageSize);

					chunkMessage.AckSessionId = (uint)new Random().Next(50000, int.MaxValue);

					chunkMessage.PrepareMessage();

					// now send it to propbably a SB processor
					try
					{
						if(MessageProcessor != null && ((SocketMessageProcessor)MessageProcessor).Connected == true)
						{
							if(DirectConnected == true)
							{
								MessageProcessor.SendMessage(new P2PDCMessage(chunkMessage));
							}
							else
							{
								// wrap the message before sending it to the (probably) SB processor
								MessageProcessor.SendMessage(WrapMessage(chunkMessage));
							}
						}

						else
						{
							InvalidateProcessor();
							BufferMessage(chunkMessage);
						}
					}
					catch(System.Net.Sockets.SocketException)
					{
						InvalidateProcessor();
						BufferMessage(chunkMessage);
					}

					bytesSend += chunkMessage.MessageSize;
				}
			}
			else
			{
				try
				{
					if(MessageProcessor != null)
					{
						if(DirectConnected == true)
						{
							MessageProcessor.SendMessage(new P2PDCMessage(p2pMessage));
						}
						else
						{
							// wrap the message before sending it to the (probably) SB processor
							MessageProcessor.SendMessage(WrapMessage(p2pMessage));
						}
					}
					else
					{
						InvalidateProcessor();
						BufferMessage(p2pMessage);
					}
				}
				catch(System.Net.Sockets.SocketException)
				{
					InvalidateProcessor();
					BufferMessage(p2pMessage);
				}
			}
		}

		/// <summary>
		/// Occurs when the processor has been marked as invalid. Due to connection error, or message processor being null.
		/// </summary>
		public event EventHandler	ProcessorInvalid;

		/// <summary>
		/// Keeps track of unsend messages
		/// </summary>
		private Queue sendMessages = new Queue();

		/// <summary>
		///
		/// </summary>
		private	bool  processorValid = true;

		/// <summary>
		/// Indicates whether the processor is invalid
		/// </summary>
		public	bool ProcessorValid
		{
			get { return processorValid; }
		}

		/// <summary>
		/// Sets the processor as invalid, and requests the p2phandler for a new request.
		/// </summary>
		protected virtual void InvalidateProcessor()
		{
			if(processorValid == false)
				return;

			processorValid = false;
			OnProcessorInvalid();

		}

		/// <summary>
		/// Sets the processor as valid.
		/// </summary>
		protected virtual void ValidateProcessor()
		{
			processorValid = true;
		}

		/// <summary>
		/// Fires the ProcessorInvalid event.
		/// </summary>
		protected virtual void OnProcessorInvalid()
		{
			if(ProcessorInvalid != null)
				ProcessorInvalid(this, new EventArgs());
		}

		/// <summary>
		/// Buffer messages that can not be send because of an invalid message processor.
		/// </summary>
		/// <param name="message"></param>
		protected virtual void BufferMessage(NetworkMessage message)
		{
			if(sendMessages.Count >= 100)
				System.Threading.Thread.Sleep(200);

			sendMessages.Enqueue(message);
		}

		/// <summary>
		/// Try to resend any messages that were stored in the buffer.
		/// </summary>
		protected virtual void SendBuffer()
		{
			if(MessageProcessor == null)
				return;

			try
			{
				while(sendMessages.Count > 0)
				{
					if(DirectConnected == true)
						MessageProcessor.SendMessage(new P2PDCMessage((P2PMessage)sendMessages.Dequeue()));
					else
						MessageProcessor.SendMessage(WrapMessage((NetworkMessage)sendMessages.Dequeue()));
				}
			}
			catch(System.Net.Sockets.SocketException)
			{
				InvalidateProcessor();
			}
		}

		private ArrayList handlers = new ArrayList();

		/// <summary>
		/// Registers a message handler. After registering the handler will receive incoming messages.
		/// </summary>
		/// <param name="handler"></param>
		public void RegisterHandler(IMessageHandler handler)
		{
			lock(handlers)
			{
				if(handlers.Contains(handler) == true)
					return;
				handlers.Add(handler);
			}
		}

		/// <summary>
		/// Unregisters a message handler. After registering the handler will no longer receive incoming messages.
		/// </summary>
		/// <param name="handler"></param>
		public void UnregisterHandler(IMessageHandler handler)
		{
			lock(handlers)
			{
				handlers.Remove(handler);
			}
		}

		#endregion
	}
}