diff --git a/ceilometer/publisher/direct.py b/ceilometer/publisher/direct.py
new file mode 100644
index 000000000..660d7932b
--- /dev/null
+++ b/ceilometer/publisher/direct.py
@@ -0,0 +1,55 @@
+#
+# Copyright 2015 Red Hat
+#
+# Author: Chris Dent <chdent@redhat.com>
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+
+
+from oslo.config import cfg
+from oslo.utils import timeutils
+
+from ceilometer.dispatcher import database
+from ceilometer import publisher
+from ceilometer.publisher import utils
+
+
+class DirectPublisher(publisher.PublisherBase):
+    """A publisher that allows saving directly from the pipeline.
+
+    Samples are saved to the currently configured database by hitching
+    a ride on the DatabaseDispatcher. This is useful where it is desirable
+    to limit the number of external services that are required.
+    """
+
+    def __init__(self, parsed_url):
+        super(DirectPublisher, self).__init__(parsed_url)
+        self.meter_conn = database.DatabaseDispatcher(cfg.CONF).meter_conn
+
+    def publish_samples(self, context, samples):
+        if not isinstance(samples, list):
+            samples = [samples]
+
+        # Transform the Sample objects into a list of dicts
+        meters = [
+            utils.meter_message_from_counter(
+                sample,
+                cfg.CONF.publisher.metering_secret)
+            for sample in samples
+        ]
+
+        for meter in meters:
+            if meter.get('timestamp'):
+                ts = timeutils.parse_isotime(meter['timestamp'])
+                meter['timestamp'] = timeutils.normalize_time(ts)
+            self.meter_conn.record_metering_data(meter)
diff --git a/ceilometer/tests/publisher/test_direct.py b/ceilometer/tests/publisher/test_direct.py
new file mode 100644
index 000000000..9cf23090f
--- /dev/null
+++ b/ceilometer/tests/publisher/test_direct.py
@@ -0,0 +1,84 @@
+#
+# Copyright 2015 Red Hat
+#
+# Author: Chris Dent <chdent@redhat.com>
+#
+# Licensed under the Apache License, Version 2.0 (the "License"); you may
+# not use this file except in compliance with the License. You may obtain
+# a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations
+# under the License.
+"""Tests for ceilometer/publisher/direct.py
+"""
+
+import datetime
+import uuid
+
+from oslo.utils import netutils
+
+from ceilometer.publisher import direct
+from ceilometer import sample
+from ceilometer.tests import db as tests_db
+
+
+class TestDirectPublisher(tests_db.TestBase,
+                          tests_db.MixinTestsWithBackendScenarios):
+
+    resource_id = str(uuid.uuid4())
+
+    test_data = [
+        sample.Sample(
+            name='alpha',
+            type=sample.TYPE_CUMULATIVE,
+            unit='',
+            volume=1,
+            user_id='test',
+            project_id='test',
+            resource_id=resource_id,
+            timestamp=datetime.datetime.utcnow().isoformat(),
+            resource_metadata={'name': 'TestPublish'},
+        ),
+        sample.Sample(
+            name='beta',
+            type=sample.TYPE_CUMULATIVE,
+            unit='',
+            volume=1,
+            user_id='test',
+            project_id='test',
+            resource_id=resource_id,
+            timestamp=datetime.datetime.utcnow().isoformat(),
+            resource_metadata={'name': 'TestPublish'},
+        ),
+        sample.Sample(
+            name='gamma',
+            type=sample.TYPE_CUMULATIVE,
+            unit='',
+            volume=1,
+            user_id='test',
+            project_id='test',
+            resource_id=resource_id,
+            timestamp=datetime.datetime.now().isoformat(),
+            resource_metadata={'name': 'TestPublish'},
+        ),
+    ]
+
+    def test_direct_publisher(self):
+        """Test samples are saved."""
+        self.CONF.set_override('connection', self.db_manager.url,
+                               group='database')
+        parsed_url = netutils.urlsplit('direct://')
+        publisher = direct.DirectPublisher(parsed_url)
+        publisher.publish_samples(None,
+                                  self.test_data)
+
+        meters = list(self.conn.get_meters(resource=self.resource_id))
+        names = sorted([meter.name for meter in meters])
+
+        self.assertEqual(3, len(meters), 'There should be 3 samples')
+        self.assertEqual(['alpha', 'beta', 'gamma'], names)
diff --git a/setup.cfg b/setup.cfg
index a51761c14..f2a5716fa 100755
--- a/setup.cfg
+++ b/setup.cfg
@@ -259,6 +259,7 @@ ceilometer.publisher =
     notifier = ceilometer.publisher.messaging:NotifierPublisher
     udp = ceilometer.publisher.udp:UDPPublisher
     file = ceilometer.publisher.file:FilePublisher
+    direct = ceilometer.publisher.direct:DirectPublisher
 
 ceilometer.alarm.evaluator =
     threshold = ceilometer.alarm.evaluator.threshold:ThresholdEvaluator