Merge lp:~pedronis/ubuntu-push/so-it-begins into lp:ubuntu-push

[Samuele Pedroni]

and as a start we get some protocol support and a development server and infrastructure

and as a start we get some protocol support and a development server and infrastructure

[John Lenton (chipaca)]

Bring it ON!

review: Approve

=== added file '.bzrignore'
--- .bzrignore	1970-01-01 00:00:00 +0000
+++ .bzrignore	2014-01-14 15:09:46 +0000
@@ -0,0 +1,3 @@
+acceptanceclient
+testserver
+coverhtml

=== added file 'COPYING'
--- COPYING	1970-01-01 00:00:00 +0000
+++ COPYING	2014-01-14 15:09:46 +0000
@@ -0,0 +1,674 @@
+                    GNU GENERAL PUBLIC LICENSE
+                       Version 3, 29 June 2007
+
+ Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>
+ Everyone is permitted to copy and distribute verbatim copies
+ of this license document, but changing it is not allowed.
+
+                            Preamble
+
+  The GNU General Public License is a free, copyleft license for
+software and other kinds of works.
+
+  The licenses for most software and other practical works are designed
+to take away your freedom to share and change the works.  By contrast,
+the GNU General Public License is intended to guarantee your freedom to
+share and change all versions of a program--to make sure it remains free
+software for all its users.  We, the Free Software Foundation, use the
+GNU General Public License for most of our software; it applies also to
+any other work released this way by its authors.  You can apply it to
+your programs, too.
+
+  When we speak of free software, we are referring to freedom, not
+price.  Our General Public Licenses are designed to make sure that you
+have the freedom to distribute copies of free software (and charge for
+them if you wish), that you receive source code or can get it if you
+want it, that you can change the software or use pieces of it in new
+free programs, and that you know you can do these things.
+
+  To protect your rights, we need to prevent others from denying you
+these rights or asking you to surrender the rights.  Therefore, you have
+certain responsibilities if you distribute copies of the software, or if
+you modify it: responsibilities to respect the freedom of others.
+
+  For example, if you distribute copies of such a program, whether
+gratis or for a fee, you must pass on to the recipients the same
+freedoms that you received.  You must make sure that they, too, receive
+or can get the source code.  And you must show them these terms so they
+know their rights.
+
+  Developers that use the GNU GPL protect your rights with two steps:
+(1) assert copyright on the software, and (2) offer you this License
+giving you legal permission to copy, distribute and/or modify it.
+
+  For the developers' and authors' protection, the GPL clearly explains
+that there is no warranty for this free software.  For both users' and
+authors' sake, the GPL requires that modified versions be marked as
+changed, so that their problems will not be attributed erroneously to
+authors of previous versions.
+
+  Some devices are designed to deny users access to install or run
+modified versions of the software inside them, although the manufacturer
+can do so.  This is fundamentally incompatible with the aim of
+protecting users' freedom to change the software.  The systematic
+pattern of such abuse occurs in the area of products for individuals to
+use, which is precisely where it is most unacceptable.  Therefore, we
+have designed this version of the GPL to prohibit the practice for those
+products.  If such problems arise substantially in other domains, we
+stand ready to extend this provision to those domains in future versions
+of the GPL, as needed to protect the freedom of users.
+
+  Finally, every program is threatened constantly by software patents.
+States should not allow patents to restrict development and use of
+software on general-purpose computers, but in those that do, we wish to
+avoid the special danger that patents applied to a free program could
+make it effectively proprietary.  To prevent this, the GPL assures that
+patents cannot be used to render the program non-free.
+
+  The precise terms and conditions for copying, distribution and
+modification follow.
+
+                       TERMS AND CONDITIONS
+
+  0. Definitions.
+
+  "This License" refers to version 3 of the GNU General Public License.
+
+  "Copyright" also means copyright-like laws that apply to other kinds of
+works, such as semiconductor masks.
+
+  "The Program" refers to any copyrightable work licensed under this
+License.  Each licensee is addressed as "you".  "Licensees" and
+"recipients" may be individuals or organizations.
+
+  To "modify" a work means to copy from or adapt all or part of the work
+in a fashion requiring copyright permission, other than the making of an
+exact copy.  The resulting work is called a "modified version" of the
+earlier work or a work "based on" the earlier work.
+
+  A "covered work" means either the unmodified Program or a work based
+on the Program.
+
+  To "propagate" a work means to do anything with it that, without
+permission, would make you directly or secondarily liable for
+infringement under applicable copyright law, except executing it on a
+computer or modifying a private copy.  Propagation includes copying,
+distribution (with or without modification), making available to the
+public, and in some countries other activities as well.
+
+  To "convey" a work means any kind of propagation that enables other
+parties to make or receive copies.  Mere interaction with a user through
+a computer network, with no transfer of a copy, is not conveying.
+
+  An interactive user interface displays "Appropriate Legal Notices"
+to the extent that it includes a convenient and prominently visible
+feature that (1) displays an appropriate copyright notice, and (2)
+tells the user that there is no warranty for the work (except to the
+extent that warranties are provided), that licensees may convey the
+work under this License, and how to view a copy of this License.  If
+the interface presents a list of user commands or options, such as a
+menu, a prominent item in the list meets this criterion.
+
+  1. Source Code.
+
+  The "source code" for a work means the preferred form of the work
+for making modifications to it.  "Object code" means any non-source
+form of a work.
+
+  A "Standard Interface" means an interface that either is an official
+standard defined by a recognized standards body, or, in the case of
+interfaces specified for a particular programming language, one that
+is widely used among developers working in that language.
+
+  The "System Libraries" of an executable work include anything, other
+than the work as a whole, that (a) is included in the normal form of
+packaging a Major Component, but which is not part of that Major
+Component, and (b) serves only to enable use of the work with that
+Major Component, or to implement a Standard Interface for which an
+implementation is available to the public in source code form.  A
+"Major Component", in this context, means a major essential component
+(kernel, window system, and so on) of the specific operating system
+(if any) on which the executable work runs, or a compiler used to
+produce the work, or an object code interpreter used to run it.
+
+  The "Corresponding Source" for a work in object code form means all
+the source code needed to generate, install, and (for an executable
+work) run the object code and to modify the work, including scripts to
+control those activities.  However, it does not include the work's
+System Libraries, or general-purpose tools or generally available free
+programs which are used unmodified in performing those activities but
+which are not part of the work.  For example, Corresponding Source
+includes interface definition files associated with source files for
+the work, and the source code for shared libraries and dynamically
+linked subprograms that the work is specifically designed to require,
+such as by intimate data communication or control flow between those
+subprograms and other parts of the work.
+
+  The Corresponding Source need not include anything that users
+can regenerate automatically from other parts of the Corresponding
+Source.
+
+  The Corresponding Source for a work in source code form is that
+same work.
+
+  2. Basic Permissions.
+
+  All rights granted under this License are granted for the term of
+copyright on the Program, and are irrevocable provided the stated
+conditions are met.  This License explicitly affirms your unlimited
+permission to run the unmodified Program.  The output from running a
+covered work is covered by this License only if the output, given its
+content, constitutes a covered work.  This License acknowledges your
+rights of fair use or other equivalent, as provided by copyright law.
+
+  You may make, run and propagate covered works that you do not
+convey, without conditions so long as your license otherwise remains
+in force.  You may convey covered works to others for the sole purpose
+of having them make modifications exclusively for you, or provide you
+with facilities for running those works, provided that you comply with
+the terms of this License in conveying all material for which you do
+not control copyright.  Those thus making or running the covered works
+for you must do so exclusively on your behalf, under your direction
+and control, on terms that prohibit them from making any copies of
+your copyrighted material outside their relationship with you.
+
+  Conveying under any other circumstances is permitted solely under
+the conditions stated below.  Sublicensing is not allowed; section 10
+makes it unnecessary.
+
+  3. Protecting Users' Legal Rights From Anti-Circumvention Law.
+
+  No covered work shall be deemed part of an effective technological
+measure under any applicable law fulfilling obligations under article
+11 of the WIPO copyright treaty adopted on 20 December 1996, or
+similar laws prohibiting or restricting circumvention of such
+measures.
+
+  When you convey a covered work, you waive any legal power to forbid
+circumvention of technological measures to the extent such circumvention
+is effected by exercising rights under this License with respect to
+the covered work, and you disclaim any intention to limit operation or
+modification of the work as a means of enforcing, against the work's
+users, your or third parties' legal rights to forbid circumvention of
+technological measures.
+
+  4. Conveying Verbatim Copies.
+
+  You may convey verbatim copies of the Program's source code as you
+receive it, in any medium, provided that you conspicuously and
+appropriately publish on each copy an appropriate copyright notice;
+keep intact all notices stating that this License and any
+non-permissive terms added in accord with section 7 apply to the code;
+keep intact all notices of the absence of any warranty; and give all
+recipients a copy of this License along with the Program.
+
+  You may charge any price or no price for each copy that you convey,
+and you may offer support or warranty protection for a fee.
+
+  5. Conveying Modified Source Versions.
+
+  You may convey a work based on the Program, or the modifications to
+produce it from the Program, in the form of source code under the
+terms of section 4, provided that you also meet all of these conditions:
+
+    a) The work must carry prominent notices stating that you modified
+    it, and giving a relevant date.
+
+    b) The work must carry prominent notices stating that it is
+    released under this License and any conditions added under section
+    7.  This requirement modifies the requirement in section 4 to
+    "keep intact all notices".
+
+    c) You must license the entire work, as a whole, under this
+    License to anyone who comes into possession of a copy.  This
+    License will therefore apply, along with any applicable section 7
+    additional terms, to the whole of the work, and all its parts,
+    regardless of how they are packaged.  This License gives no
+    permission to license the work in any other way, but it does not
+    invalidate such permission if you have separately received it.
+
+    d) If the work has interactive user interfaces, each must display
+    Appropriate Legal Notices; however, if the Program has interactive
+    interfaces that do not display Appropriate Legal Notices, your
+    work need not make them do so.
+
+  A compilation of a covered work with other separate and independent
+works, which are not by their nature extensions of the covered work,
+and which are not combined with it such as to form a larger program,
+in or on a volume of a storage or distribution medium, is called an
+"aggregate" if the compilation and its resulting copyright are not
+used to limit the access or legal rights of the compilation's users
+beyond what the individual works permit.  Inclusion of a covered work
+in an aggregate does not cause this License to apply to the other
+parts of the aggregate.
+
+  6. Conveying Non-Source Forms.
+
+  You may convey a covered work in object code form under the terms
+of sections 4 and 5, provided that you also convey the
+machine-readable Corresponding Source under the terms of this License,
+in one of these ways:
+
+    a) Convey the object code in, or embodied in, a physical product
+    (including a physical distribution medium), accompanied by the
+    Corresponding Source fixed on a durable physical medium
+    customarily used for software interchange.
+
+    b) Convey the object code in, or embodied in, a physical product
+    (including a physical distribution medium), accompanied by a
+    written offer, valid for at least three years and valid for as
+    long as you offer spare parts or customer support for that product
+    model, to give anyone who possesses the object code either (1) a
+    copy of the Corresponding Source for all the software in the
+    product that is covered by this License, on a durable physical
+    medium customarily used for software interchange, for a price no
+    more than your reasonable cost of physically performing this
+    conveying of source, or (2) access to copy the
+    Corresponding Source from a network server at no charge.
+
+    c) Convey individual copies of the object code with a copy of the
+    written offer to provide the Corresponding Source.  This
+    alternative is allowed only occasionally and noncommercially, and
+    only if you received the object code with such an offer, in accord
+    with subsection 6b.
+
+    d) Convey the object code by offering access from a designated
+    place (gratis or for a charge), and offer equivalent access to the
+    Corresponding Source in the same way through the same place at no
+    further charge.  You need not require recipients to copy the
+    Corresponding Source along with the object code.  If the place to
+    copy the object code is a network server, the Corresponding Source
+    may be on a different server (operated by you or a third party)
+    that supports equivalent copying facilities, provided you maintain
+    clear directions next to the object code saying where to find the
+    Corresponding Source.  Regardless of what server hosts the
+    Corresponding Source, you remain obligated to ensure that it is
+    available for as long as needed to satisfy these requirements.
+
+    e) Convey the object code using peer-to-peer transmission, provided
+    you inform other peers where the object code and Corresponding
+    Source of the work are being offered to the general public at no
+    charge under subsection 6d.
+
+  A separable portion of the object code, whose source code is excluded
+from the Corresponding Source as a System Library, need not be
+included in conveying the object code work.
+
+  A "User Product" is either (1) a "consumer product", which means any
+tangible personal property which is normally used for personal, family,
+or household purposes, or (2) anything designed or sold for incorporation
+into a dwelling.  In determining whether a product is a consumer product,
+doubtful cases shall be resolved in favor of coverage.  For a particular
+product received by a particular user, "normally used" refers to a
+typical or common use of that class of product, regardless of the status
+of the particular user or of the way in which the particular user
+actually uses, or expects or is expected to use, the product.  A product
+is a consumer product regardless of whether the product has substantial
+commercial, industrial or non-consumer uses, unless such uses represent
+the only significant mode of use of the product.
+
+  "Installation Information" for a User Product means any methods,
+procedures, authorization keys, or other information required to install
+and execute modified versions of a covered work in that User Product from
+a modified version of its Corresponding Source.  The information must
+suffice to ensure that the continued functioning of the modified object
+code is in no case prevented or interfered with solely because
+modification has been made.
+
+  If you convey an object code work under this section in, or with, or
+specifically for use in, a User Product, and the conveying occurs as
+part of a transaction in which the right of possession and use of the
+User Product is transferred to the recipient in perpetuity or for a
+fixed term (regardless of how the transaction is characterized), the
+Corresponding Source conveyed under this section must be accompanied
+by the Installation Information.  But this requirement does not apply
+if neither you nor any third party retains the ability to install
+modified object code on the User Product (for example, the work has
+been installed in ROM).
+
+  The requirement to provide Installation Information does not include a
+requirement to continue to provide support service, warranty, or updates
+for a work that has been modified or installed by the recipient, or for
+the User Product in which it has been modified or installed.  Access to a
+network may be denied when the modification itself materially and
+adversely affects the operation of the network or violates the rules and
+protocols for communication across the network.
+
+  Corresponding Source conveyed, and Installation Information provided,
+in accord with this section must be in a format that is publicly
+documented (and with an implementation available to the public in
+source code form), and must require no special password or key for
+unpacking, reading or copying.
+
+  7. Additional Terms.
+
+  "Additional permissions" are terms that supplement the terms of this
+License by making exceptions from one or more of its conditions.
+Additional permissions that are applicable to the entire Program shall
+be treated as though they were included in this License, to the extent
+that they are valid under applicable law.  If additional permissions
+apply only to part of the Program, that part may be used separately
+under those permissions, but the entire Program remains governed by
+this License without regard to the additional permissions.
+
+  When you convey a copy of a covered work, you may at your option
+remove any additional permissions from that copy, or from any part of
+it.  (Additional permissions may be written to require their own
+removal in certain cases when you modify the work.)  You may place
+additional permissions on material, added by you to a covered work,
+for which you have or can give appropriate copyright permission.
+
+  Notwithstanding any other provision of this License, for material you
+add to a covered work, you may (if authorized by the copyright holders of
+that material) supplement the terms of this License with terms:
+
+    a) Disclaiming warranty or limiting liability differently from the
+    terms of sections 15 and 16 of this License; or
+
+    b) Requiring preservation of specified reasonable legal notices or
+    author attributions in that material or in the Appropriate Legal
+    Notices displayed by works containing it; or
+
+    c) Prohibiting misrepresentation of the origin of that material, or
+    requiring that modified versions of such material be marked in
+    reasonable ways as different from the original version; or
+
+    d) Limiting the use for publicity purposes of names of licensors or
+    authors of the material; or
+
+    e) Declining to grant rights under trademark law for use of some
+    trade names, trademarks, or service marks; or
+
+    f) Requiring indemnification of licensors and authors of that
+    material by anyone who conveys the material (or modified versions of
+    it) with contractual assumptions of liability to the recipient, for
+    any liability that these contractual assumptions directly impose on
+    those licensors and authors.
+
+  All other non-permissive additional terms are considered "further
+restrictions" within the meaning of section 10.  If the Program as you
+received it, or any part of it, contains a notice stating that it is
+governed by this License along with a term that is a further
+restriction, you may remove that term.  If a license document contains
+a further restriction but permits relicensing or conveying under this
+License, you may add to a covered work material governed by the terms
+of that license document, provided that the further restriction does
+not survive such relicensing or conveying.
+
+  If you add terms to a covered work in accord with this section, you
+must place, in the relevant source files, a statement of the
+additional terms that apply to those files, or a notice indicating
+where to find the applicable terms.
+
+  Additional terms, permissive or non-permissive, may be stated in the
+form of a separately written license, or stated as exceptions;
+the above requirements apply either way.
+
+  8. Termination.
+
+  You may not propagate or modify a covered work except as expressly
+provided under this License.  Any attempt otherwise to propagate or
+modify it is void, and will automatically terminate your rights under
+this License (including any patent licenses granted under the third
+paragraph of section 11).
+
+  However, if you cease all violation of this License, then your
+license from a particular copyright holder is reinstated (a)
+provisionally, unless and until the copyright holder explicitly and
+finally terminates your license, and (b) permanently, if the copyright
+holder fails to notify you of the violation by some reasonable means
+prior to 60 days after the cessation.
+
+  Moreover, your license from a particular copyright holder is
+reinstated permanently if the copyright holder notifies you of the
+violation by some reasonable means, this is the first time you have
+received notice of violation of this License (for any work) from that
+copyright holder, and you cure the violation prior to 30 days after
+your receipt of the notice.
+
+  Termination of your rights under this section does not terminate the
+licenses of parties who have received copies or rights from you under
+this License.  If your rights have been terminated and not permanently
+reinstated, you do not qualify to receive new licenses for the same
+material under section 10.
+
+  9. Acceptance Not Required for Having Copies.
+
+  You are not required to accept this License in order to receive or
+run a copy of the Program.  Ancillary propagation of a covered work
+occurring solely as a consequence of using peer-to-peer transmission
+to receive a copy likewise does not require acceptance.  However,
+nothing other than this License grants you permission to propagate or
+modify any covered work.  These actions infringe copyright if you do
+not accept this License.  Therefore, by modifying or propagating a
+covered work, you indicate your acceptance of this License to do so.
+
+  10. Automatic Licensing of Downstream Recipients.
+
+  Each time you convey a covered work, the recipient automatically
+receives a license from the original licensors, to run, modify and
+propagate that work, subject to this License.  You are not responsible
+for enforcing compliance by third parties with this License.
+
+  An "entity transaction" is a transaction transferring control of an
+organization, or substantially all assets of one, or subdividing an
+organization, or merging organizations.  If propagation of a covered
+work results from an entity transaction, each party to that
+transaction who receives a copy of the work also receives whatever
+licenses to the work the party's predecessor in interest had or could
+give under the previous paragraph, plus a right to possession of the
+Corresponding Source of the work from the predecessor in interest, if
+the predecessor has it or can get it with reasonable efforts.
+
+  You may not impose any further restrictions on the exercise of the
+rights granted or affirmed under this License.  For example, you may
+not impose a license fee, royalty, or other charge for exercise of
+rights granted under this License, and you may not initiate litigation
+(including a cross-claim or counterclaim in a lawsuit) alleging that
+any patent claim is infringed by making, using, selling, offering for
+sale, or importing the Program or any portion of it.
+
+  11. Patents.
+
+  A "contributor" is a copyright holder who authorizes use under this
+License of the Program or a work on which the Program is based.  The
+work thus licensed is called the contributor's "contributor version".
+
+  A contributor's "essential patent claims" are all patent claims
+owned or controlled by the contributor, whether already acquired or
+hereafter acquired, that would be infringed by some manner, permitted
+by this License, of making, using, or selling its contributor version,
+but do not include claims that would be infringed only as a
+consequence of further modification of the contributor version.  For
+purposes of this definition, "control" includes the right to grant
+patent sublicenses in a manner consistent with the requirements of
+this License.
+
+  Each contributor grants you a non-exclusive, worldwide, royalty-free
+patent license under the contributor's essential patent claims, to
+make, use, sell, offer for sale, import and otherwise run, modify and
+propagate the contents of its contributor version.
+
+  In the following three paragraphs, a "patent license" is any express
+agreement or commitment, however denominated, not to enforce a patent
+(such as an express permission to practice a patent or covenant not to
+sue for patent infringement).  To "grant" such a patent license to a
+party means to make such an agreement or commitment not to enforce a
+patent against the party.
+
+  If you convey a covered work, knowingly relying on a patent license,
+and the Corresponding Source of the work is not available for anyone
+to copy, free of charge and under the terms of this License, through a
+publicly available network server or other readily accessible means,
+then you must either (1) cause the Corresponding Source to be so
+available, or (2) arrange to deprive yourself of the benefit of the
+patent license for this particular work, or (3) arrange, in a manner
+consistent with the requirements of this License, to extend the patent
+license to downstream recipients.  "Knowingly relying" means you have
+actual knowledge that, but for the patent license, your conveying the
+covered work in a country, or your recipient's use of the covered work
+in a country, would infringe one or more identifiable patents in that
+country that you have reason to believe are valid.
+
+  If, pursuant to or in connection with a single transaction or
+arrangement, you convey, or propagate by procuring conveyance of, a
+covered work, and grant a patent license to some of the parties
+receiving the covered work authorizing them to use, propagate, modify
+or convey a specific copy of the covered work, then the patent license
+you grant is automatically extended to all recipients of the covered
+work and works based on it.
+
+  A patent license is "discriminatory" if it does not include within
+the scope of its coverage, prohibits the exercise of, or is
+conditioned on the non-exercise of one or more of the rights that are
+specifically granted under this License.  You may not convey a covered
+work if you are a party to an arrangement with a third party that is
+in the business of distributing software, under which you make payment
+to the third party based on the extent of your activity of conveying
+the work, and under which the third party grants, to any of the
+parties who would receive the covered work from you, a discriminatory
+patent license (a) in connection with copies of the covered work
+conveyed by you (or copies made from those copies), or (b) primarily
+for and in connection with specific products or compilations that
+contain the covered work, unless you entered into that arrangement,
+or that patent license was granted, prior to 28 March 2007.
+
+  Nothing in this License shall be construed as excluding or limiting
+any implied license or other defenses to infringement that may
+otherwise be available to you under applicable patent law.
+
+  12. No Surrender of Others' Freedom.
+
+  If conditions are imposed on you (whether by court order, agreement or
+otherwise) that contradict the conditions of this License, they do not
+excuse you from the conditions of this License.  If you cannot convey a
+covered work so as to satisfy simultaneously your obligations under this
+License and any other pertinent obligations, then as a consequence you may
+not convey it at all.  For example, if you agree to terms that obligate you
+to collect a royalty for further conveying from those to whom you convey
+the Program, the only way you could satisfy both those terms and this
+License would be to refrain entirely from conveying the Program.
+
+  13. Use with the GNU Affero General Public License.
+
+  Notwithstanding any other provision of this License, you have
+permission to link or combine any covered work with a work licensed
+under version 3 of the GNU Affero General Public License into a single
+combined work, and to convey the resulting work.  The terms of this
+License will continue to apply to the part which is the covered work,
+but the special requirements of the GNU Affero General Public License,
+section 13, concerning interaction through a network will apply to the
+combination as such.
+
+  14. Revised Versions of this License.
+
+  The Free Software Foundation may publish revised and/or new versions of
+the GNU General Public License from time to time.  Such new versions will
+be similar in spirit to the present version, but may differ in detail to
+address new problems or concerns.
+
+  Each version is given a distinguishing version number.  If the
+Program specifies that a certain numbered version of the GNU General
+Public License "or any later version" applies to it, you have the
+option of following the terms and conditions either of that numbered
+version or of any later version published by the Free Software
+Foundation.  If the Program does not specify a version number of the
+GNU General Public License, you may choose any version ever published
+by the Free Software Foundation.
+
+  If the Program specifies that a proxy can decide which future
+versions of the GNU General Public License can be used, that proxy's
+public statement of acceptance of a version permanently authorizes you
+to choose that version for the Program.
+
+  Later license versions may give you additional or different
+permissions.  However, no additional obligations are imposed on any
+author or copyright holder as a result of your choosing to follow a
+later version.
+
+  15. Disclaimer of Warranty.
+
+  THERE IS NO WARRANTY FOR THE PROGRAM, TO THE EXTENT PERMITTED BY
+APPLICABLE LAW.  EXCEPT WHEN OTHERWISE STATED IN WRITING THE COPYRIGHT
+HOLDERS AND/OR OTHER PARTIES PROVIDE THE PROGRAM "AS IS" WITHOUT WARRANTY
+OF ANY KIND, EITHER EXPRESSED OR IMPLIED, INCLUDING, BUT NOT LIMITED TO,
+THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+PURPOSE.  THE ENTIRE RISK AS TO THE QUALITY AND PERFORMANCE OF THE PROGRAM
+IS WITH YOU.  SHOULD THE PROGRAM PROVE DEFECTIVE, YOU ASSUME THE COST OF
+ALL NECESSARY SERVICING, REPAIR OR CORRECTION.
+
+  16. Limitation of Liability.
+
+  IN NO EVENT UNLESS REQUIRED BY APPLICABLE LAW OR AGREED TO IN WRITING
+WILL ANY COPYRIGHT HOLDER, OR ANY OTHER PARTY WHO MODIFIES AND/OR CONVEYS
+THE PROGRAM AS PERMITTED ABOVE, BE LIABLE TO YOU FOR DAMAGES, INCLUDING ANY
+GENERAL, SPECIAL, INCIDENTAL OR CONSEQUENTIAL DAMAGES ARISING OUT OF THE
+USE OR INABILITY TO USE THE PROGRAM (INCLUDING BUT NOT LIMITED TO LOSS OF
+DATA OR DATA BEING RENDERED INACCURATE OR LOSSES SUSTAINED BY YOU OR THIRD
+PARTIES OR A FAILURE OF THE PROGRAM TO OPERATE WITH ANY OTHER PROGRAMS),
+EVEN IF SUCH HOLDER OR OTHER PARTY HAS BEEN ADVISED OF THE POSSIBILITY OF
+SUCH DAMAGES.
+
+  17. Interpretation of Sections 15 and 16.
+
+  If the disclaimer of warranty and limitation of liability provided
+above cannot be given local legal effect according to their terms,
+reviewing courts shall apply local law that most closely approximates
+an absolute waiver of all civil liability in connection with the
+Program, unless a warranty or assumption of liability accompanies a
+copy of the Program in return for a fee.
+
+                     END OF TERMS AND CONDITIONS
+
+            How to Apply These Terms to Your New Programs
+
+  If you develop a new program, and you want it to be of the greatest
+possible use to the public, the best way to achieve this is to make it
+free software which everyone can redistribute and change under these terms.
+
+  To do so, attach the following notices to the program.  It is safest
+to attach them to the start of each source file to most effectively
+state the exclusion of warranty; and each file should have at least
+the "copyright" line and a pointer to where the full notice is found.
+
+    <one line to give the program's name and a brief idea of what it does.>
+    Copyright (C) <year>  <name of author>
+
+    This program is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    This program is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+Also add information on how to contact you by electronic and paper mail.
+
+  If the program does terminal interaction, make it output a short
+notice like this when it starts in an interactive mode:
+
+    <program>  Copyright (C) <year>  <name of author>
+    This program comes with ABSOLUTELY NO WARRANTY; for details type `show w'.
+    This is free software, and you are welcome to redistribute it
+    under certain conditions; type `show c' for details.
+
+The hypothetical commands `show w' and `show c' should show the appropriate
+parts of the General Public License.  Of course, your program's commands
+might be different; for a GUI interface, you would use an "about box".
+
+  You should also get your employer (if you work as a programmer) or school,
+if any, to sign a "copyright disclaimer" for the program, if necessary.
+For more information on this, and how to apply and follow the GNU GPL, see
+<http://www.gnu.org/licenses/>.
+
+  The GNU General Public License does not permit incorporating your program
+into proprietary programs.  If your program is a subroutine library, you
+may consider it more useful to permit linking proprietary applications with
+the library.  If this is what you want to do, use the GNU Lesser General
+Public License instead of this License.  But first, please read
+<http://www.gnu.org/philosophy/why-not-lgpl.html>.

=== added file 'LICENSE'
--- LICENSE	1970-01-01 00:00:00 +0000
+++ LICENSE	2014-01-14 15:09:46 +0000
@@ -0,0 +1,16 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+

=== added file 'Makefile'
--- Makefile	1970-01-01 00:00:00 +0000
+++ Makefile	2014-01-14 15:09:46 +0000
@@ -0,0 +1,48 @@
+GOPATH := $(shell cd ../../..; pwd)
+export GOPATH
+
+PROJECT = launchpad.net/ubuntu-push
+
+ifneq ($(CURDIR),$(GOPATH)/src/launchpad.net/ubuntu-push)
+$(error unexpected curdir and/or layout)
+endif
+
+GODEPS = launchpad.net/gocheck
+
+bootstrap:
+	mkdir -p $(GOPATH)/bin
+	mkdir -p $(GOPATH)/pkg
+	go get -u launchpad.net/godeps
+	go get -d -u $(GODEPS)
+	$(GOPATH)/bin/godeps -u dependencies.tsv
+	go install $(GODEPS)
+
+check:
+	go test $(PROJECT)/...
+
+check-race:
+	go test -race $(PROJECT)/...
+
+coverage-summary:
+	go test -a -cover $(PROJECT)/...
+
+coverage-html:
+	mkdir -p coverhtml
+	for pkg in $$(go list $(PROJECT)/...|grep -v acceptance ); do \
+		relname="$${pkg#$(PROJECT)/}" ; \
+		mkdir -p coverhtml/$$(dirname $${relname}) ; \
+		go test -a -coverprofile=coverhtml/$${relname}.out $$pkg ; \
+		if [ -f coverhtml/$${relname}.out ] ; then \
+	           go tool cover -html=coverhtml/$${relname}.out -o coverhtml/$${relname}.html ; \
+	           go tool cover -func=coverhtml/$${relname}.out -o coverhtml/$${relname}.txt ; \
+		fi \
+	done
+
+format:
+	go fmt $(PROJECT)/...
+
+check-format:
+	scripts/check_fmt $(PROJECT)
+
+.PHONY: bootstrap check check-race format check-format coverage-summary \
+	coverage-html

=== added file 'README'
--- README	1970-01-01 00:00:00 +0000
+++ README	2014-01-14 15:09:46 +0000
@@ -0,0 +1,37 @@
+Ubuntu Push Notifications
+--------------------------
+
+Protocol, client, and development code for Ubuntu Push Notifications.
+
+The code expects to be checked out as launchpad.net/ubuntu-push in a Go
+workspace, see go help gopath.
+
+To setup go dependencies one can use:
+
+   make bootstrap
+
+To run tests:
+
+   make check
+
+To produce coverage reports go 1.2 (default on trusty) is needed and
+the cover tool (the latter can be obtained atm with something like:
+sudo GOPATH=<go-workspace> go get code.google.com/p/go.tools/cmd/cover
+)
+
+then run:
+
+   make coverage-summary
+
+for a summary report, or:
+
+for per package html with annotated code in coverhtml/<package-name>.html
+
+  make coverage-html
+
+(it makes also textual coverhtml/<package-name>.txt reports).
+
+To run the acceptance tests, go into the acceptance subdir and run:
+
+   ./acceptance.sh
+

=== added directory 'config'
=== added file 'config/config.go'
--- config/config.go	1970-01-01 00:00:00 +0000
+++ config/config.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,149 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// Package config has helpers to parse and use JSON based configuration.
+package config
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net"
+	"path/filepath"
+	"reflect"
+	"strings"
+	"time"
+)
+
+// ReadConfig reads a JSON configuration into destConfig which should
+// be a pointer to a structure, it does some more configuration
+// specific error checking than plain JSON decoding and mentions
+// fields in errors . Configuration fields are expected to start with
+// lower case in the JSON object.
+func ReadConfig(r io.Reader, destConfig interface{}) error {
+	destValue := reflect.ValueOf(destConfig)
+	if destValue.Kind() != reflect.Ptr || destValue.Elem().Kind() != reflect.Struct {
+		return errors.New("destConfig not *struct")
+	}
+	// do the parsing in two phases for better error handling
+	var p1 map[string]json.RawMessage
+	err := json.NewDecoder(r).Decode(&p1)
+	if err != nil {
+		return err
+	}
+	destStruct := destValue.Elem()
+	structType := destStruct.Type()
+	n := structType.NumField()
+	for i := 0; i < n; i++ {
+		fld := structType.Field(i)
+		configName := strings.Split(fld.Tag.Get("json"), ",")[0]
+		if configName == "" {
+			configName = strings.ToLower(fld.Name[:1]) + fld.Name[1:]
+		}
+		raw, found := p1[configName]
+		if !found { // assume all fields are mandatory for now
+			return fmt.Errorf("missing %s", configName)
+		}
+		dest := destStruct.Field(i).Addr().Interface()
+		err = json.Unmarshal([]byte(raw), dest)
+		if err != nil {
+			return fmt.Errorf("%s: %v", configName, err)
+		}
+	}
+	return nil
+}
+
+// ConfigTimeDuration can hold a time.Duration in a configuration struct,
+// that is parsed from a string as supported by time.ParseDuration.
+type ConfigTimeDuration struct {
+	time.Duration
+}
+
+func (ctd *ConfigTimeDuration) UnmarshalJSON(b []byte) error {
+	var enc string
+	var v time.Duration
+	err := json.Unmarshal(b, &enc)
+	if err != nil {
+		return err
+	}
+	v, err = time.ParseDuration(enc)
+	if err != nil {
+		return err
+	}
+	*ctd = ConfigTimeDuration{v}
+	return nil
+}
+
+// TimeDuration returns the time.Duration held in ctd.
+func (ctd ConfigTimeDuration) TimeDuration() time.Duration {
+	return ctd.Duration
+}
+
+// ConfigHostPort can hold a host:port string in a configuration struct.
+type ConfigHostPort string
+
+func (chp *ConfigHostPort) UnmarshalJSON(b []byte) error {
+	var enc string
+	err := json.Unmarshal(b, &enc)
+	if err != nil {
+		return err
+	}
+	_, _, err = net.SplitHostPort(enc)
+	if err != nil {
+		return err
+	}
+	*chp = ConfigHostPort(enc)
+	return nil
+}
+
+// HostPort returns the host:port string held in chp.
+func (chp ConfigHostPort) HostPort() string {
+	return string(chp)
+}
+
+// ConfigQueueSize can hold a queue size in a configuration struct.
+type ConfigQueueSize uint
+
+func (cqs *ConfigQueueSize) UnmarshalJSON(b []byte) error {
+	var enc uint
+	err := json.Unmarshal(b, &enc)
+	if err != nil {
+		return err
+	}
+	if enc == 0 {
+		return errors.New("queue size should be > 0")
+	}
+	*cqs = ConfigQueueSize(enc)
+	return nil
+}
+
+// QueueSize returns the queue size held in cqs.
+func (cqs ConfigQueueSize) QueueSize() uint {
+	return uint(cqs)
+}
+
+// LoadFile reads  a file possibly relative to a base dir.
+func LoadFile(p, baseDir string) ([]byte, error) {
+	if p == "" {
+		return nil, nil
+	}
+	if !filepath.IsAbs(p) {
+		p = filepath.Join(baseDir, p)
+	}
+	return ioutil.ReadFile(p)
+}

=== added file 'config/config_test.go'
--- config/config_test.go	1970-01-01 00:00:00 +0000
+++ config/config_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,135 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package config
+
+import (
+	"bytes"
+	"io/ioutil"
+	. "launchpad.net/gocheck"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestConfig(t *testing.T) { TestingT(t) }
+
+type configSuite struct{}
+
+var _ = Suite(&configSuite{})
+
+type testConfig1 struct {
+	A int
+	B string
+	C []string `json:"c_list"`
+}
+
+func (s *configSuite) TestReadConfig(c *C) {
+	buf := bytes.NewBufferString(`{"a": 1, "b": "foo", "c_list": ["c", "d", "e"]}`)
+	var cfg testConfig1
+	err := ReadConfig(buf, &cfg)
+	c.Check(err, IsNil)
+	c.Check(cfg, DeepEquals, testConfig1{A: 1, B: "foo", C: []string{"c", "d", "e"}})
+}
+
+func checkError(c *C, config string, dest interface{}, expectedError string) {
+	buf := bytes.NewBufferString(config)
+	err := ReadConfig(buf, dest)
+	c.Check(err, ErrorMatches, expectedError)
+}
+
+func (s *configSuite) TestReadConfigErrors(c *C) {
+	var cfg testConfig1
+	checkError(c, "", cfg, `destConfig not \*struct`)
+	var i int
+	checkError(c, "", &i, `destConfig not \*struct`)
+	checkError(c, "", &cfg, `EOF`)
+	checkError(c, `{"a": "1"}`, &cfg, `a: .*type int`)
+	checkError(c, `{"b": "1"}`, &cfg, `missing a`)
+	checkError(c, `{"A": "1"}`, &cfg, `missing a`)
+	checkError(c, `{"a": 1, "b": "foo"}`, &cfg, `missing c_list`)
+}
+
+type testTimeDurationConfig struct {
+	D ConfigTimeDuration
+}
+
+func (s *configSuite) TestReadConfigTimeDuration(c *C) {
+	buf := bytes.NewBufferString(`{"d": "2s"}`)
+	var cfg testTimeDurationConfig
+	err := ReadConfig(buf, &cfg)
+	c.Assert(err, IsNil)
+	c.Check(cfg.D.TimeDuration(), Equals, 2*time.Second)
+}
+
+func (s *configSuite) TestReadConfigTimeDurationErrors(c *C) {
+	var cfg testTimeDurationConfig
+	checkError(c, `{"d": 1}`, &cfg, "d:.*type string")
+	checkError(c, `{"d": "2"}`, &cfg, "d:.*missing unit.*")
+}
+
+type testHostPortConfig struct {
+	H ConfigHostPort
+}
+
+func (s *configSuite) TestReadConfigHostPort(c *C) {
+	buf := bytes.NewBufferString(`{"h": "127.0.0.1:9999"}`)
+	var cfg testHostPortConfig
+	err := ReadConfig(buf, &cfg)
+	c.Assert(err, IsNil)
+	c.Check(cfg.H.HostPort(), Equals, "127.0.0.1:9999")
+}
+
+func (s *configSuite) TestReadConfigHostPortErrors(c *C) {
+	var cfg testHostPortConfig
+	checkError(c, `{"h": 1}`, &cfg, "h:.*type string")
+	checkError(c, `{"h": ""}`, &cfg, "h: missing port in address")
+}
+
+type testQueueSizeConfig struct {
+	QS ConfigQueueSize
+}
+
+func (s *configSuite) TestReadConfigQueueSize(c *C) {
+	buf := bytes.NewBufferString(`{"qS": 1}`)
+	var cfg testQueueSizeConfig
+	err := ReadConfig(buf, &cfg)
+	c.Assert(err, IsNil)
+	c.Check(cfg.QS.QueueSize(), Equals, uint(1))
+}
+
+func (s *configSuite) TestReadConfigQueueSizeErrors(c *C) {
+	var cfg testQueueSizeConfig
+	checkError(c, `{"qS": "x"}`, &cfg, "qS: .*type uint")
+	checkError(c, `{"qS": 0}`, &cfg, "qS: queue size should be > 0")
+}
+
+func (s *configSuite) TestLoadFile(c *C) {
+	tmpDir := c.MkDir()
+	d, err := LoadFile("", tmpDir)
+	c.Check(err, IsNil)
+	c.Check(d, IsNil)
+	fullPath := filepath.Join(tmpDir, "example.file")
+	err = ioutil.WriteFile(fullPath, []byte("Example"), os.ModePerm)
+	c.Assert(err, IsNil)
+	d, err = LoadFile("example.file", tmpDir)
+	c.Check(err, IsNil)
+	c.Check(string(d), Equals, "Example")
+	d, err = LoadFile(fullPath, tmpDir)
+	c.Check(err, IsNil)
+	c.Check(string(d), Equals, "Example")
+}

=== added file 'dependencies.tsv'
--- dependencies.tsv	1970-01-01 00:00:00 +0000
+++ dependencies.tsv	2014-01-14 15:09:46 +0000
@@ -0,0 +1,1 @@
+launchpad.net/gocheck	bzr	[e-mail]-20130302024745-6ikofwq2c03h7giu	85

=== added directory 'logger'
=== added file 'logger/logger.go'
--- logger/logger.go	1970-01-01 00:00:00 +0000
+++ logger/logger.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,100 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// Package logger defines a simple logger API with level of logging control.
+package logger
+
+import (
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"runtime"
+)
+
+// Logger is a simple logger interface with logging at levels.
+type Logger interface {
+	// Errorf logs an error.
+	Errorf(format string, v ...interface{})
+	// Fatalf logs an error and exists the program with os.Exit(1).
+	Fatalf(format string, v ...interface{})
+	// Recoverf recover from a possible panic and logs it.
+	Recoverf(format string, v ...interface{})
+	// Infof logs a info message.
+	Infof(format string, v ...interface{})
+	// Debugf logs a debug message.
+	Debugf(format string, v ...interface{})
+}
+
+type simpleLogger struct {
+	*log.Logger
+	nlevel int
+}
+
+const (
+	lError = iota
+	lInfo
+	lDebug
+)
+
+var levelToNLevel = map[string]int{
+	"error": lError,
+	"info":  lInfo,
+	"debug": lDebug,
+}
+
+// NewSimpleLogger creates a logger logging only up to the given level.
+// level can be in order: "error", "info", "debug".
+func NewSimpleLogger(w io.Writer, level string) Logger {
+	nlevel := levelToNLevel[level]
+	return &simpleLogger{
+		log.New(w, "", log.Ldate|log.Ltime|log.Lmicroseconds),
+		nlevel,
+	}
+}
+
+func (lg *simpleLogger) Errorf(format string, v ...interface{}) {
+	lg.Printf("ERROR "+format, v...)
+}
+
+var osExit = os.Exit // for testing
+
+func (lg *simpleLogger) Fatalf(format string, v ...interface{}) {
+	lg.Printf("ERROR "+format, v...)
+	osExit(1)
+}
+
+func (lg *simpleLogger) Recoverf(format string, v ...interface{}) {
+	if err := recover(); err != nil {
+		msg := fmt.Sprintf(format, v...)
+		stack := make([]byte, 8*1024) // Stack writes less but doesn't fail
+		stackWritten := runtime.Stack(stack, false)
+		stack = stack[:stackWritten]
+		lg.Printf("ERROR panic %v!! %s:\n%s", err, msg, stack)
+	}
+}
+
+func (lg *simpleLogger) Infof(format string, v ...interface{}) {
+	if lg.nlevel >= lInfo {
+		lg.Printf("INFO "+format, v...)
+	}
+}
+
+func (lg *simpleLogger) Debugf(format string, v ...interface{}) {
+	if lg.nlevel >= lDebug {
+		lg.Printf("DEBUG "+format, v...)
+	}
+}

=== added file 'logger/logger_test.go'
--- logger/logger_test.go	1970-01-01 00:00:00 +0000
+++ logger/logger_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,117 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package logger
+
+import (
+	"bytes"
+	. "launchpad.net/gocheck"
+	"os"
+	"testing"
+)
+
+func TestLogger(t *testing.T) { TestingT(t) }
+
+type loggerSuite struct{}
+
+var _ = Suite(&loggerSuite{})
+
+func (s *loggerSuite) TestErrorf(c *C) {
+	buf := &bytes.Buffer{}
+	logger := NewSimpleLogger(buf, "error")
+	logger.Errorf("%v %d", "error", 1)
+	c.Check(buf.String(), Matches, ".* ERROR error 1\n")
+}
+
+func (s *loggerSuite) TestFatalf(c *C) {
+	defer func() {
+		osExit = os.Exit
+	}()
+	var exitCode int
+	osExit = func(code int) {
+		exitCode = code
+	}
+	buf := &bytes.Buffer{}
+	logger := NewSimpleLogger(buf, "error")
+	logger.Fatalf("%v %v", "error", "fatal")
+	c.Check(buf.String(), Matches, ".* ERROR error fatal\n")
+	c.Check(exitCode, Equals, 1)
+}
+
+func (s *loggerSuite) TestInfof(c *C) {
+	buf := &bytes.Buffer{}
+	logger := NewSimpleLogger(buf, "info")
+	logger.Infof("%v %d", "info", 1)
+	c.Check(buf.String(), Matches, ".* INFO info 1\n")
+}
+
+func (s *loggerSuite) TestDebugf(c *C) {
+	buf := &bytes.Buffer{}
+	logger := NewSimpleLogger(buf, "debug")
+	logger.Debugf("%v %d", "debug", 1)
+	c.Check(buf.String(), Matches, `.* DEBUG debug 1\n`)
+}
+
+func (s *loggerSuite) TestFormat(c *C) {
+	buf := &bytes.Buffer{}
+	logger := NewSimpleLogger(buf, "error")
+	logger.Errorf("%v %d", "error", 2)
+	c.Check(buf.String(), Matches, `.* .*\.\d+ ERROR error 2\n`)
+}
+
+func (s *loggerSuite) TestLevel(c *C) {
+	buf := &bytes.Buffer{}
+	logger := NewSimpleLogger(buf, "error")
+	logger.Errorf("%s%d", "e", 3)
+	logger.Infof("%s%d", "i", 3)
+	logger.Debugf("%s%d", "d", 3)
+	c.Check(buf.String(), Matches, `.* ERROR e3\n`)
+
+	buf.Reset()
+	logger = NewSimpleLogger(buf, "info")
+	logger.Errorf("%s%d", "e", 4)
+	logger.Debugf("%s%d", "d", 4)
+	logger.Infof("%s%d", "i", 4)
+	c.Check(buf.String(), Matches, `.* ERROR e4\n.* INFO i4\n`)
+
+	buf.Reset()
+	logger = NewSimpleLogger(buf, "debug")
+	logger.Errorf("%s%d", "e", 5)
+	logger.Debugf("%s%d", "d", 5)
+	logger.Infof("%s%d", "i", 5)
+	c.Check(buf.String(), Matches, `.* ERROR e5\n.* DEBUG d5\n.* INFO i5\n`)
+}
+
+func panicAndRecover(logger Logger, n int, doPanic bool) {
+	defer logger.Recoverf("%v %d", "panic", n)
+	if doPanic {
+		panic("Troubles")
+	}
+}
+
+func (s *loggerSuite) TestRecoverf(c *C) {
+	buf := &bytes.Buffer{}
+	logger := NewSimpleLogger(buf, "error")
+	panicAndRecover(logger, 6, true)
+	c.Check(buf.String(), Matches, "(?s).* ERROR panic Troubles!! panic 6:.*panicAndRecover.*")
+}
+
+func (s *loggerSuite) TestRecoverfNop(c *C) {
+	buf := &bytes.Buffer{}
+	logger := NewSimpleLogger(buf, "error")
+	panicAndRecover(logger, 6, false)
+	c.Check(buf.String(), Equals, "")
+}

=== added directory 'protocol'
=== added file 'protocol/messages.go'
--- protocol/messages.go	1970-01-01 00:00:00 +0000
+++ protocol/messages.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,103 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package protocol
+
+// representative struct for messages
+
+import (
+	"encoding/json"
+	// "log"
+)
+
+//  System channel id using a shortened hex encoded form for the NIL UUID
+const SystemChannelId = "0"
+
+// CONNECT message
+type ConnectMsg struct {
+	Type      string `json:"T"`
+	ClientVer string
+	DeviceId  string
+	Info      map[string]interface{} `json:",omitempty"` // platform etc...
+	// maps channel ids (hex encoded UUIDs) to known client channel levels
+	Levels map[string]int64
+}
+
+// PING/PONG messages
+type PingPongMsg struct {
+	Type string `json:"T"`
+}
+
+const maxPayloadSize = 62 * 1024
+
+// SplittableMsg are messages that may require and are capable of splitting.
+type SplittableMsg interface {
+	Split() (done bool)
+}
+
+// BROADCAST messages
+type BroadcastMsg struct {
+	Type      string `json:"T"`
+	AppId     string `json:",omitempty"`
+	ChanId    string
+	TopLevel  int64
+	Payloads  []json.RawMessage
+	splitting int
+}
+
+func (m *BroadcastMsg) Split() bool {
+	var prevTop int64
+	if m.splitting == 0 {
+		prevTop = m.TopLevel - int64(len(m.Payloads))
+	} else {
+		prevTop = m.TopLevel
+		m.Payloads = m.Payloads[len(m.Payloads):m.splitting]
+		m.TopLevel = prevTop + int64(len(m.Payloads))
+	}
+	payloads := m.Payloads
+	var size int
+	for i := range payloads {
+		size += len(payloads[i])
+		if size > maxPayloadSize {
+			m.TopLevel = prevTop + int64(i)
+			m.splitting = len(payloads)
+			m.Payloads = payloads[:i]
+			return false
+		}
+	}
+	return true
+}
+
+// NOTIFICATIONS message
+type NotificationsMsg struct {
+	Type          string `json:"T"`
+	Notifications []Notification
+}
+
+// A single unicast notification
+type Notification struct {
+	AppId string `json:"A"`
+	MsgId string `json:"M"`
+	// payload
+	Payload json.RawMessage `json:"P"`
+}
+
+// ACKnowelgement message
+type AckMsg struct {
+	Type string `json:"T"`
+}
+
+// xxx ... query levels messages

=== added file 'protocol/messages_test.go'
--- protocol/messages_test.go	1970-01-01 00:00:00 +0000
+++ protocol/messages_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,101 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package protocol
+
+import (
+	"encoding/json"
+	"fmt"
+	. "launchpad.net/gocheck"
+	"strings"
+)
+
+type messagesSuite struct{}
+
+var _ = Suite(&messagesSuite{})
+
+func (s *messagesSuite) TestSplitBroadcastMsgNop(c *C) {
+	b := &BroadcastMsg{
+		Type:     "broadcast",
+		AppId:    "APP",
+		ChanId:   "0",
+		TopLevel: 2,
+		Payloads: []json.RawMessage{json.RawMessage(`{b:1}`), json.RawMessage(`{b:2}`)},
+	}
+	done := b.Split()
+	c.Check(done, Equals, true)
+	c.Check(b.TopLevel, Equals, int64(2))
+	c.Check(cap(b.Payloads), Equals, 2)
+	c.Check(len(b.Payloads), Equals, 2)
+}
+
+var payloadFmt = fmt.Sprintf(`{"b":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
+
+func manyParts(c int) []json.RawMessage {
+	payloads := make([]json.RawMessage, 0, 1)
+	for i := 0; i < c; i++ {
+		payloads = append(payloads, json.RawMessage(fmt.Sprintf(payloadFmt, i)))
+	}
+	return payloads
+}
+
+func (s *messagesSuite) TestSplitBroadcastMsgManyParts(c *C) {
+	payloads := manyParts(33)
+	n := len(payloads)
+	// more interesting this way
+	c.Assert(cap(payloads), Not(Equals), n)
+	b := &BroadcastMsg{
+		Type:     "broadcast",
+		AppId:    "APP",
+		ChanId:   "0",
+		TopLevel: 500,
+		Payloads: payloads,
+	}
+	done := b.Split()
+	c.Assert(done, Equals, false)
+	n1 := len(b.Payloads)
+	c.Check(b.TopLevel, Equals, int64(500-n+n1))
+	buf, err := json.Marshal(b)
+	c.Assert(err, IsNil)
+	c.Assert(len(buf) <= 65535, Equals, true)
+	c.Check(len(buf)+len(payloads[n1]) > maxPayloadSize, Equals, true)
+	done = b.Split()
+	c.Assert(done, Equals, true)
+	n2 := len(b.Payloads)
+	c.Check(b.TopLevel, Equals, int64(500))
+	c.Check(n1+n2, Equals, n)
+
+	payloads = manyParts(61)
+	n = len(payloads)
+	b = &BroadcastMsg{
+		Type:     "broadcast",
+		AppId:    "APP",
+		ChanId:   "0",
+		TopLevel: int64(n),
+		Payloads: payloads,
+	}
+	done = b.Split()
+	c.Assert(done, Equals, false)
+	n1 = len(b.Payloads)
+	done = b.Split()
+	c.Assert(done, Equals, false)
+	n2 = len(b.Payloads)
+	done = b.Split()
+	c.Assert(done, Equals, true)
+	n3 := len(b.Payloads)
+	c.Check(b.TopLevel, Equals, int64(n))
+	c.Check(n1+n2+n3, Equals, n)
+}

=== added file 'protocol/protocol.go'
--- protocol/protocol.go	1970-01-01 00:00:00 +0000
+++ protocol/protocol.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,103 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// Package protocol has code to talk the client-daemon<->push-server protocol.
+package protocol
+
+import (
+	"bytes"
+	"encoding/binary"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net"
+	"time"
+)
+
+// Protocol is a connection capable of writing and reading the wire format of protocol messages.
+type Protocol interface {
+	SetDeadline(t time.Time)
+	ReadMessage(msg interface{}) error
+	WriteMessage(msg interface{}) error
+}
+
+func ReadWireFormatVersion(conn net.Conn, exchangeTimeout time.Duration) (ver int, err error) {
+	var buf1 [1]byte
+	err = conn.SetReadDeadline(time.Now().Add(exchangeTimeout))
+	if err != nil {
+		panic(fmt.Errorf("can't set deadline: %v", err))
+	}
+	_, err = conn.Read(buf1[:])
+	ver = int(buf1[0])
+	return
+}
+
+const ProtocolWireVersion = 0
+
+// protocol0 handles version 0 of the wire format
+type protocol0 struct {
+	buffer *bytes.Buffer
+	enc    *json.Encoder
+	conn   net.Conn
+}
+
+// NewProtocol0 creates and initialises a protocol with wire format version 0.
+func NewProtocol0(conn net.Conn) *protocol0 {
+	buf := bytes.NewBuffer(make([]byte, 5000))
+	return &protocol0{
+		buffer: buf,
+		enc:    json.NewEncoder(buf),
+		conn:   conn}
+}
+
+// SetDeadline sets deadline for the subsquent WriteMessage/ReadMessage exchange
+func (c *protocol0) SetDeadline(t time.Time) {
+	err := c.conn.SetDeadline(t)
+	if err != nil {
+		panic(fmt.Errorf("can't set deadline: %v", err))
+	}
+}
+
+// ReadMessage reads one message made of big endian uint16 length, JSON body of length from the connection.
+func (c *protocol0) ReadMessage(msg interface{}) error {
+	c.buffer.Reset()
+	_, err := io.CopyN(c.buffer, c.conn, 2)
+	if err != nil {
+		return err
+	}
+	length := binary.BigEndian.Uint16(c.buffer.Bytes())
+	c.buffer.Reset()
+	_, err = io.CopyN(c.buffer, c.conn, int64(length))
+	if err != nil {
+		return err
+	}
+	return json.Unmarshal(c.buffer.Bytes(), msg)
+}
+
+// WriteMessage writes one message made of big endian uint16 length, JSON body of length to the connection.
+func (c *protocol0) WriteMessage(msg interface{}) error {
+	c.buffer.Reset()
+	c.buffer.WriteString("\x00\x00") // placeholder for length
+	err := c.enc.Encode(msg)
+	if err != nil {
+		panic(fmt.Errorf("WriteMessage got: %v", err))
+	}
+	msgLen := c.buffer.Len() - 3 // length space, extra newline
+	toWrite := c.buffer.Bytes()
+	binary.BigEndian.PutUint16(toWrite[:2], uint16(msgLen))
+	_, err = c.conn.Write(toWrite[:msgLen+2])
+	return err
+}

=== added file 'protocol/protocol_test.go'
--- protocol/protocol_test.go	1970-01-01 00:00:00 +0000
+++ protocol/protocol_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,227 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package protocol
+
+import (
+	"encoding/binary"
+	"encoding/json"
+	// "fmt"
+	"io"
+	. "launchpad.net/gocheck"
+	"net"
+	"testing"
+	"time"
+)
+
+func TestProtocol(t *testing.T) { TestingT(t) }
+
+type protocolSuite struct{}
+
+var _ = Suite(&protocolSuite{})
+
+type deadline struct {
+	kind      string
+	deadAfter time.Duration
+}
+
+func (d *deadline) setDeadAfter(t time.Time) {
+	deadAfter := t.Sub(time.Now())
+	d.deadAfter = (deadAfter + time.Millisecond/2) / time.Millisecond * time.Millisecond
+}
+
+type rw struct {
+	buf []byte
+	n   int
+	err error
+}
+
+type testConn struct {
+	deadlines []*deadline
+	reads     []rw
+	writes    []*rw
+}
+
+func (tc *testConn) LocalAddr() net.Addr {
+	return nil
+}
+
+func (tc *testConn) RemoteAddr() net.Addr {
+	return nil
+}
+
+func (tc *testConn) Close() error {
+	return nil
+}
+
+func (tc *testConn) SetDeadline(t time.Time) error {
+	deadline := tc.deadlines[0]
+	deadline.kind = "both"
+	deadline.setDeadAfter(t)
+	tc.deadlines = tc.deadlines[1:]
+	return nil
+}
+
+func (tc *testConn) SetReadDeadline(t time.Time) error {
+	deadline := tc.deadlines[0]
+	deadline.kind = "read"
+	deadline.setDeadAfter(t)
+	tc.deadlines = tc.deadlines[1:]
+	return nil
+}
+
+func (tc *testConn) SetWriteDeadline(t time.Time) error {
+	deadline := tc.deadlines[0]
+	deadline.kind = "write"
+	deadline.setDeadAfter(t)
+	tc.deadlines = tc.deadlines[1:]
+	return nil
+}
+
+func (tc *testConn) Read(buf []byte) (n int, err error) {
+	read := tc.reads[0]
+	copy(buf, read.buf)
+	tc.reads = tc.reads[1:]
+	return read.n, read.err
+}
+
+func (tc *testConn) Write(buf []byte) (n int, err error) {
+	write := tc.writes[0]
+	n = copy(write.buf, buf)
+	write.buf = write.buf[:n]
+	write.n = n
+	err = write.err
+	tc.writes = tc.writes[1:]
+	return
+}
+
+func (s *protocolSuite) TestReadWireFormatVersion(c *C) {
+	deadl := deadline{}
+	read1 := rw{buf: []byte{42}, n: 1}
+	tc := &testConn{reads: []rw{read1}, deadlines: []*deadline{&deadl}}
+	ver, err := ReadWireFormatVersion(tc, time.Minute)
+	c.Check(err, IsNil)
+	c.Check(ver, Equals, 42)
+	c.Check(deadl.kind, Equals, "read")
+	c.Check(deadl.deadAfter, Equals, time.Minute)
+}
+
+func (s *protocolSuite) TestReadWireFormatVersionError(c *C) {
+	deadl := deadline{}
+	read1 := rw{err: io.EOF}
+	tc := &testConn{reads: []rw{read1}, deadlines: []*deadline{&deadl}}
+	_, err := ReadWireFormatVersion(tc, time.Minute)
+	c.Check(err, Equals, io.EOF)
+}
+
+func (s *protocolSuite) TestSetDeadline(c *C) {
+	deadl := deadline{}
+	tc := &testConn{deadlines: []*deadline{&deadl}}
+	pc := NewProtocol0(tc)
+	pc.SetDeadline(time.Now().Add(time.Minute))
+	c.Check(deadl.kind, Equals, "both")
+	c.Check(deadl.deadAfter, Equals, time.Minute)
+}
+
+type testMsg struct {
+	Type string `json:"T"`
+	A    uint64
+}
+
+func lengthAsBytes(length uint16) []byte {
+	var buf [2]byte
+	var res = buf[:]
+	binary.BigEndian.PutUint16(res, length)
+	return res
+}
+
+func (s *protocolSuite) TestReadMessage(c *C) {
+	msgBuf, _ := json.Marshal(testMsg{Type: "msg", A: 2000})
+	readMsgLen := rw{buf: lengthAsBytes(uint16(len(msgBuf))), n: 2}
+	readMsgBody := rw{buf: msgBuf, n: len(msgBuf)}
+	tc := &testConn{reads: []rw{readMsgLen, readMsgBody}}
+	pc := NewProtocol0(tc)
+	var recvMsg testMsg
+	err := pc.ReadMessage(&recvMsg)
+	c.Check(err, IsNil)
+	c.Check(recvMsg, DeepEquals, testMsg{Type: "msg", A: 2000})
+}
+
+func (s *protocolSuite) TestReadMessageBits(c *C) {
+	msgBuf, _ := json.Marshal(testMsg{Type: "msg", A: 2000})
+	readMsgLen := rw{buf: lengthAsBytes(uint16(len(msgBuf))), n: 2}
+	readMsgBody1 := rw{buf: msgBuf[:5], n: 5}
+	readMsgBody2 := rw{buf: msgBuf[5:], n: len(msgBuf) - 5}
+	tc := &testConn{reads: []rw{readMsgLen, readMsgBody1, readMsgBody2}}
+	pc := NewProtocol0(tc)
+	var recvMsg testMsg
+	err := pc.ReadMessage(&recvMsg)
+	c.Check(err, IsNil)
+	c.Check(recvMsg, DeepEquals, testMsg{Type: "msg", A: 2000})
+}
+
+func (s *protocolSuite) TestReadMessageIOErrors(c *C) {
+	msgBuf, _ := json.Marshal(testMsg{Type: "msg", A: 2000})
+	readMsgLenErr := rw{n: 1, err: io.ErrClosedPipe}
+	tc1 := &testConn{reads: []rw{readMsgLenErr}}
+	pc1 := NewProtocol0(tc1)
+	var recvMsg testMsg
+	err := pc1.ReadMessage(&recvMsg)
+	c.Check(err, Equals, io.ErrClosedPipe)
+
+	readMsgLen := rw{buf: lengthAsBytes(uint16(len(msgBuf))), n: 2}
+	readMsgBody1 := rw{buf: msgBuf[:5], n: 5}
+	readMsgBody2Err := rw{n: 2, err: io.EOF}
+	tc2 := &testConn{reads: []rw{readMsgLen, readMsgBody1, readMsgBody2Err}}
+	pc2 := NewProtocol0(tc2)
+	err = pc2.ReadMessage(&recvMsg)
+	c.Check(err, Equals, io.EOF)
+}
+
+func (s *protocolSuite) TestReadMessageBrokenJSON(c *C) {
+	msgBuf := []byte("{\"T\"}")
+	readMsgLen := rw{buf: lengthAsBytes(uint16(len(msgBuf))), n: 2}
+	readMsgBody := rw{buf: msgBuf, n: len(msgBuf)}
+	tc := &testConn{reads: []rw{readMsgLen, readMsgBody}}
+	pc := NewProtocol0(tc)
+	var recvMsg testMsg
+	err := pc.ReadMessage(&recvMsg)
+	c.Check(err, FitsTypeOf, &json.SyntaxError{})
+}
+
+func (s *protocolSuite) TestWriteMessage(c *C) {
+	writeMsg := rw{buf: make([]byte, 64)}
+	tc := &testConn{writes: []*rw{&writeMsg}}
+	pc := NewProtocol0(tc)
+	msg := testMsg{Type: "m", A: 9999}
+	err := pc.WriteMessage(&msg)
+	c.Check(err, IsNil)
+	var msgLen int = int(binary.BigEndian.Uint16(writeMsg.buf[:2]))
+	c.Check(msgLen, Equals, len(writeMsg.buf)-2)
+	var wroteMsg testMsg
+	formatErr := json.Unmarshal(writeMsg.buf[2:], &wroteMsg)
+	c.Check(formatErr, IsNil)
+	c.Check(wroteMsg, DeepEquals, testMsg{Type: "m", A: 9999})
+}
+
+func (s *protocolSuite) TestWriteMessageIOErrors(c *C) {
+	writeMsgErr := rw{buf: make([]byte, 0), err: io.ErrClosedPipe}
+	tc1 := &testConn{writes: []*rw{&writeMsgErr}}
+	pc1 := NewProtocol0(tc1)
+	msg := testMsg{Type: "m", A: 9999}
+	err := pc1.WriteMessage(&msg)
+	c.Check(err, Equals, io.ErrClosedPipe)
+}

=== added directory 'scripts'
=== added file 'scripts/check_fmt'
--- scripts/check_fmt	1970-01-01 00:00:00 +0000
+++ scripts/check_fmt	2014-01-14 15:09:46 +0000
@@ -0,0 +1,14 @@
+#!/bin/bash
+# check that all go files respect gofmt formatting
+# assumes GOPATH is properly set
+PROJECT=${1:?missing project}
+PROBLEMS=
+for pkg in $(go list ${PROJECT}/...) ; do
+    NONCOMPLIANT=$(gofmt -l ${GOPATH}/src/${pkg}/*.go)
+    if [ -n "${NONCOMPLIANT}" ]; then
+        echo pkg $pkg has some gofmt non-compliant files:
+        echo ${NONCOMPLIANT}|xargs -d ' ' -n1 basename
+        PROBLEMS="y"
+    fi
+done
+test -z "${PROBLEMS}"

=== added directory 'server'
=== added directory 'server/acceptance'
=== added file 'server/acceptance/acceptance.sh'
--- server/acceptance/acceptance.sh	1970-01-01 00:00:00 +0000
+++ server/acceptance/acceptance.sh	2014-01-14 15:09:46 +0000
@@ -0,0 +1,6 @@
+# run acceptance tests, expects properly setup GOPATH and deps
+# can set extra build params like -race with BUILD_FLAGS envvar
+set -ex
+go test $BUILD_FLAGS -i
+go build $BUILD_FLAGS -o testserver launchpad.net/ubuntu-push/server/dev
+go test $BUILD_FLAGS -server ./testserver $*

=== added file 'server/acceptance/acceptance_test.go'
--- server/acceptance/acceptance_test.go	1970-01-01 00:00:00 +0000
+++ server/acceptance/acceptance_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,487 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package acceptance
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"flag"
+	"fmt"
+	"io/ioutil"
+	. "launchpad.net/gocheck"
+	"launchpad.net/ubuntu-push/protocol"
+	"launchpad.net/ubuntu-push/server/api"
+	"net"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"regexp"
+	"runtime"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestAcceptance(t *testing.T) { TestingT(t) }
+
+type acceptanceSuite struct {
+	server       *exec.Cmd
+	serverAddr   string
+	serverURL    string
+	serverEvents <-chan string
+	httpClient   *http.Client
+}
+
+var _ = Suite(&acceptanceSuite{})
+
+var serverCmd = flag.String("server", "", "server to test")
+
+// SourceRelative produces a path relative to the source code, makes
+// sense only for tests when the code is available on disk.
+// xxx later move it to a testing helpers package
+func SourceRelative(relativePath string) string {
+	_, file, _, ok := runtime.Caller(1)
+	if !ok {
+		panic("failed to get source filename using Caller()")
+	}
+	return filepath.Join(filepath.Dir(file), relativePath)
+}
+
+func testServerConfig(addr, httpAddr string) map[string]interface{} {
+	cfg := map[string]interface{}{
+		"exchange_timeout":   "0.1s",
+		"ping_interval":      "0.5s",
+		"session_queue_size": 10,
+		"broker_queue_size":  100,
+		"addr":               addr,
+		"key_pem_file":       SourceRelative("config/testing.key"),
+		"cert_pem_file":      SourceRelative("config/testing.cert"),
+		"http_addr":          httpAddr,
+		"http_read_timeout":  "1s",
+		"http_write_timeout": "1s",
+	}
+	return cfg
+}
+
+func testClientSession(addr string, deviceId string, reportPings bool) *ClientSession {
+	certPEMBlock, err := ioutil.ReadFile(SourceRelative("config/testing.cert"))
+	if err != nil {
+		panic(fmt.Sprintf("could not read config/testing.cert: %v", err))
+	}
+	return &ClientSession{
+		ExchangeTimeout: 100 * time.Millisecond,
+		PingInterval:    500 * time.Millisecond,
+		ServerAddr:      addr,
+		CertPEMBlock:    certPEMBlock,
+		DeviceId:        deviceId,
+		ReportPings:     reportPings,
+	}
+}
+
+const (
+	devListeningOnPat  = "INFO listening for devices on "
+	httpListeningOnPat = "INFO listening for http on "
+)
+
+var rxLineInfo = regexp.MustCompile("^.*? ([[:alpha:]].*)\n")
+
+func extractListeningAddr(c *C, pat, line string) string {
+	if !strings.HasPrefix(line, pat) {
+		c.Fatalf("server: %v", line)
+	}
+	return line[len(pat):]
+}
+
+// start a new server for each test
+func (s *acceptanceSuite) SetUpTest(c *C) {
+	if *serverCmd == "" {
+		c.Skip("executable server not specified")
+	}
+	tmpDir := c.MkDir()
+	cfgFilename := filepath.Join(tmpDir, "config.json")
+	cfgJson, err := json.Marshal(testServerConfig("127.0.0.1:0", "127.0.0.1:0"))
+	if err != nil {
+		c.Fatal(err)
+	}
+	err = ioutil.WriteFile(cfgFilename, cfgJson, os.ModePerm)
+	if err != nil {
+		c.Fatal(err)
+	}
+	server := exec.Command(*serverCmd, cfgFilename)
+	stderr, err := server.StderrPipe()
+	if err != nil {
+		c.Fatal(err)
+	}
+	err = server.Start()
+	if err != nil {
+		c.Fatal(err)
+	}
+	bufErr := bufio.NewReaderSize(stderr, 5000)
+	getLineInfo := func() (string, error) {
+		line, err := bufErr.ReadString('\n')
+		if err != nil {
+			return "", err
+		}
+		extracted := rxLineInfo.FindStringSubmatch(line)
+		if extracted == nil {
+			return "", fmt.Errorf("unexpected server line: %#v", line)
+		}
+		return extracted[1], nil
+	}
+	infoHTTP, err := getLineInfo()
+	if err != nil {
+		c.Fatal(err)
+	}
+	serverHTTPAddr := extractListeningAddr(c, httpListeningOnPat, infoHTTP)
+	s.serverURL = fmt.Sprintf("http://%s", serverHTTPAddr)
+	info, err := getLineInfo()
+	if err != nil {
+		c.Fatal(err)
+	}
+	s.serverAddr = extractListeningAddr(c, devListeningOnPat, info)
+	s.server = server
+	serverEvents := make(chan string, 5)
+	s.serverEvents = serverEvents
+	go func() {
+		for {
+			info, err := getLineInfo()
+			if err != nil {
+				serverEvents <- fmt.Sprintf("ERROR: %v", err)
+				close(serverEvents)
+				return
+			}
+			serverEvents <- info
+		}
+	}()
+	s.httpClient = &http.Client{}
+}
+
+func (s *acceptanceSuite) TearDownTest(c *C) {
+	if s.server != nil {
+		s.server.Process.Kill()
+		s.server = nil
+	}
+}
+
+// nextEvent receives an event from given channel with a 5s timeout
+func nextEvent(events <-chan string, errCh <-chan error) string {
+	select {
+	case <-time.After(5 * time.Second):
+		panic("too long stuck waiting for next event")
+	case err := <-errCh:
+		return err.Error() // will fail comparison typically
+	case evStr := <-events:
+		return evStr
+	}
+}
+
+// Tests about connection, ping-pong, disconnection scenarios
+
+// typically combined with -gocheck.vv or test selection
+var logTraffic = flag.Bool("logTraffic", false, "log traffic")
+
+type connInterceptor func(ic *interceptingConn, op string, b []byte) (bool, int, error)
+
+type interceptingConn struct {
+	net.Conn
+	totalRead    int
+	totalWritten int
+	intercept    connInterceptor
+}
+
+func (ic *interceptingConn) Write(b []byte) (n int, err error) {
+	done := false
+	before := ic.totalWritten
+	if ic.intercept != nil {
+		done, n, err = ic.intercept(ic, "write", b)
+	}
+	if !done {
+		n, err = ic.Conn.Write(b)
+	}
+	ic.totalWritten += n
+	if *logTraffic {
+		fmt.Printf("W[%v]: %d %#v %v %d\n", ic.Conn.LocalAddr(), before, string(b[:n]), err, ic.totalWritten)
+	}
+	return
+}
+
+func (ic *interceptingConn) Read(b []byte) (n int, err error) {
+	done := false
+	before := ic.totalRead
+	if ic.intercept != nil {
+		done, n, err = ic.intercept(ic, "read", b)
+	}
+	if !done {
+		n, err = ic.Conn.Read(b)
+	}
+	ic.totalRead += n
+	if *logTraffic {
+		fmt.Printf("R[%v]: %d %#v %v %d\n", ic.Conn.LocalAddr(), before, string(b[:n]), err, ic.totalRead)
+	}
+	return
+}
+
+func (s *acceptanceSuite) TestConnectPingPing(c *C) {
+	errCh := make(chan error, 1)
+	events := make(chan string, 10)
+	sess := testClientSession(s.serverAddr, "DEVA", true)
+	err := sess.Dial()
+	c.Assert(err, IsNil)
+	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
+		// would be 3rd ping read, based on logged traffic
+		if op == "read" && ic.totalRead >= 28 {
+			// exit the sess.Run() goroutine, client will close
+			runtime.Goexit()
+		}
+		return false, 0, nil
+	}
+	sess.Connection = &interceptingConn{sess.Connection, 0, 0, intercept}
+	go func() {
+		errCh <- sess.Run(events)
+	}()
+	connectCli := nextEvent(events, errCh)
+	connectSrv := nextEvent(s.serverEvents, nil)
+	registeredSrv := nextEvent(s.serverEvents, nil)
+	tconnect := time.Now()
+	c.Assert(connectSrv, Matches, ".*session.* connected .*")
+	c.Assert(registeredSrv, Matches, ".*session.* registered DEVA")
+	c.Assert(strings.HasSuffix(connectSrv, connectCli), Equals, true)
+	c.Assert(nextEvent(events, errCh), Equals, "Ping")
+	elapsedOfPing := float64(time.Since(tconnect)) / float64(500*time.Millisecond)
+	c.Check(elapsedOfPing >= 1.0, Equals, true)
+	c.Check(elapsedOfPing < 1.05, Equals, true)
+	c.Assert(nextEvent(events, errCh), Equals, "Ping")
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, ".*session.* ended with: EOF")
+	c.Check(len(errCh), Equals, 0)
+}
+
+func (s *acceptanceSuite) TestConnectPingNeverPong(c *C) {
+	errCh := make(chan error, 1)
+	events := make(chan string, 10)
+	sess := testClientSession(s.serverAddr, "DEVB", true)
+	err := sess.Dial()
+	c.Assert(err, IsNil)
+	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
+		// would be pong to 2nd ping, based on logged traffic
+		if op == "write" && ic.totalRead >= 28 {
+			time.Sleep(200 * time.Millisecond)
+			// exit the sess.Run() goroutine, client will close
+			runtime.Goexit()
+		}
+		return false, 0, nil
+	}
+	sess.Connection = &interceptingConn{sess.Connection, 0, 0, intercept}
+	go func() {
+		errCh <- sess.Run(events)
+	}()
+	c.Assert(nextEvent(events, errCh), Matches, "connected .*")
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, ".*session.* connected .*")
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, ".*session.* registered .*")
+	c.Assert(nextEvent(events, errCh), Equals, "Ping")
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*timeout`)
+	c.Check(len(errCh), Equals, 0)
+}
+
+// Tests about broadcast
+
+func (s *acceptanceSuite) postRequest(path string, message interface{}) (string, error) {
+	packedMessage, err := json.Marshal(message)
+	if err != nil {
+		panic(err)
+	}
+	reader := bytes.NewReader(packedMessage)
+
+	url := s.serverURL + path
+	request, _ := http.NewRequest("POST", url, reader)
+	request.ContentLength = int64(reader.Len())
+	request.Header.Set("Content-Type", "application/json")
+
+	resp, err := s.httpClient.Do(request)
+	if err != nil {
+		panic(err)
+	}
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	return string(body), err
+}
+
+func (s *acceptanceSuite) startClient(c *C, devId string, intercept connInterceptor, levels map[string]int64) (<-chan string, <-chan error) {
+	errCh := make(chan error, 1)
+	events := make(chan string, 10)
+	sess := testClientSession(s.serverAddr, devId, false)
+	sess.Levels = levels
+	err := sess.Dial()
+	c.Assert(err, IsNil)
+	sess.Connection = &interceptingConn{sess.Connection, 0, 0, intercept}
+	go func() {
+		errCh <- sess.Run(events)
+	}()
+	c.Assert(nextEvent(events, errCh), Matches, "connected .*")
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, ".*session.* connected .*")
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, ".*session.* registered "+devId)
+	return events, errCh
+}
+
+func (s *acceptanceSuite) TestBroadcastToConnected(c *C) {
+	clientShutdown := make(chan bool, 1) // abused as an atomic flag
+	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
+		// read after ack
+		if op == "read" && len(clientShutdown) > 0 {
+			// exit the sess.Run() goroutine, client will close
+			runtime.Goexit()
+		}
+		return false, 0, nil
+	}
+	events, errCh := s.startClient(c, "DEVB", intercept, nil)
+	got, err := s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"n": 42}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+	c.Check(nextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"n":42}]`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}
+
+func (s *acceptanceSuite) TestBroadcastPending(c *C) {
+	// send broadcast that will be pending
+	got, err := s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"b": 1}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+
+	clientShutdown := make(chan bool, 1) // abused as an atomic flag
+	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
+		// read after ack
+		if op == "read" && len(clientShutdown) > 0 {
+			// exit the sess.Run() goroutine, client will close
+			runtime.Goexit()
+		}
+		return false, 0, nil
+	}
+	events, errCh := s.startClient(c, "DEVB", intercept, nil)
+	// gettting pending on connect
+	c.Check(nextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"b":1}]`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}
+
+func (s *acceptanceSuite) TestBroadcasLargeNeedsSplitting(c *C) {
+	// send bunch of broadcasts that will be pending
+	payloadFmt := fmt.Sprintf(`{"b":%%d,"bloat":"%s"}`, strings.Repeat("x", 1024*2))
+	for i := 0; i < 32; i++ {
+		got, err := s.postRequest("/broadcast", &api.Broadcast{
+			Channel: "system",
+			Data:    json.RawMessage(fmt.Sprintf(payloadFmt, i)),
+		})
+		c.Check(err, IsNil)
+		c.Check(got, Matches, ".*ok.*")
+	}
+
+	clientShutdown := make(chan bool, 1) // abused as an atomic flag
+	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
+		// read after ack
+		if op == "read" && len(clientShutdown) > 0 {
+			// exit the sess.Run() goroutine, client will close
+			runtime.Goexit()
+		}
+		return false, 0, nil
+	}
+	events, errCh := s.startClient(c, "DEVC", intercept, nil)
+	// gettting pending on connect
+	c.Check(nextEvent(events, errCh), Matches, `broadcast chan:0 app: topLevel:30 payloads:\[{"b":0,.*`)
+	c.Check(nextEvent(events, errCh), Matches, `broadcast chan:0 app: topLevel:32 payloads:\[.*`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}
+
+func (s *acceptanceSuite) TestBroadcastDistribution2(c *C) {
+	clientShutdown := make(chan bool, 1) // abused as an atomic flag
+	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
+		// read after ack
+		if op == "read" && len(clientShutdown) > 0 {
+			// exit the sess.Run() goroutine, client will close
+			runtime.Goexit()
+		}
+		return false, 0, nil
+	}
+	// start 1st clinet
+	events1, errCh1 := s.startClient(c, "DEV1", intercept, nil)
+	// start 2nd client
+	events2, errCh2 := s.startClient(c, "DEV2", intercept, nil)
+	// broadcast
+	got, err := s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"n": 42}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+	c.Check(nextEvent(events1, errCh1), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"n":42}]`)
+	c.Check(nextEvent(events2, errCh2), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"n":42}]`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh1), Equals, 0)
+	c.Check(len(errCh2), Equals, 0)
+}
+
+func (s *acceptanceSuite) TestBroadcastFilterByLevel(c *C) {
+	clientShutdown := make(chan bool, 1) // abused as an atomic flag
+	intercept := func(ic *interceptingConn, op string, b []byte) (bool, int, error) {
+		// read after ack
+		if op == "read" && len(clientShutdown) > 0 {
+			// exit the sess.Run() goroutine, client will close
+			runtime.Goexit()
+		}
+		return false, 0, nil
+	}
+	events, errCh := s.startClient(c, "DEVD", intercept, nil)
+	got, err := s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"b": 1}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+	c.Check(nextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:1 payloads:[{"b":1}]`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+	// another broadcast
+	got, err = s.postRequest("/broadcast", &api.Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"b": 2}`),
+	})
+	c.Check(err, IsNil)
+	c.Check(got, Matches, ".*ok.*")
+	// reconnect, provide levels, get only later notification
+	<-clientShutdown // reset
+	events, errCh = s.startClient(c, "DEVD", intercept, map[string]int64{
+		protocol.SystemChannelId: 1,
+	})
+	c.Check(nextEvent(events, errCh), Equals, `broadcast chan:0 app: topLevel:2 payloads:[{"b":2}]`)
+	clientShutdown <- true
+	c.Assert(nextEvent(s.serverEvents, nil), Matches, `.* ended with:.*EOF`)
+	c.Check(len(errCh), Equals, 0)
+}

=== added file 'server/acceptance/acceptanceclient.go'
--- server/acceptance/acceptanceclient.go	1970-01-01 00:00:00 +0000
+++ server/acceptance/acceptanceclient.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,123 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package acceptance
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"launchpad.net/ubuntu-push/protocol"
+	"net"
+	"time"
+)
+
+var wireVersionBytes = []byte{protocol.ProtocolWireVersion}
+
+// ClienSession holds a client<->server session and its configuration.
+type ClientSession struct {
+	// configuration
+	DeviceId        string
+	ServerAddr      string
+	PingInterval    time.Duration
+	ExchangeTimeout time.Duration
+	CertPEMBlock    []byte
+	ReportPings     bool
+	Levels          map[string]int64
+	// connection
+	Connection net.Conn
+}
+
+// Dial connects to a server using the configuration in the ClientSession
+// and sets up the connection.
+func (sess *ClientSession) Dial() error {
+	conn, err := net.DialTimeout("tcp", sess.ServerAddr, sess.ExchangeTimeout)
+	if err != nil {
+		return err
+	}
+	tlsConfig := &tls.Config{}
+	if sess.CertPEMBlock != nil {
+		cp := x509.NewCertPool()
+		ok := cp.AppendCertsFromPEM(sess.CertPEMBlock)
+		if !ok {
+			return errors.New("dial: could not parse certificate")
+		}
+		tlsConfig.RootCAs = cp
+	}
+	sess.Connection = tls.Client(conn, tlsConfig)
+	return nil
+}
+
+type serverMsg struct {
+	Type string `json:"T"`
+	protocol.BroadcastMsg
+	protocol.NotificationsMsg
+}
+
+// Run the session with the server, emits a stream of events.
+func (sess *ClientSession) Run(events chan<- string) error {
+	conn := sess.Connection
+	defer conn.Close()
+	conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+	_, err := conn.Write(wireVersionBytes)
+	if err != nil {
+		return err
+	}
+	proto := protocol.NewProtocol0(conn)
+	err = proto.WriteMessage(protocol.ConnectMsg{
+		Type:     "connect",
+		DeviceId: sess.DeviceId,
+		Levels:   sess.Levels,
+	})
+	if err != nil {
+		return err
+	}
+	events <- fmt.Sprintf("connected %v", conn.LocalAddr())
+	var recv serverMsg
+	for {
+		deadAfter := sess.PingInterval + sess.ExchangeTimeout
+		conn.SetDeadline(time.Now().Add(deadAfter))
+		err = proto.ReadMessage(&recv)
+		if err != nil {
+			return err
+		}
+		switch recv.Type {
+		case "ping":
+			conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+			err := proto.WriteMessage(protocol.PingPongMsg{Type: "pong"})
+			if err != nil {
+				return err
+			}
+			if sess.ReportPings {
+				events <- "Ping"
+			}
+		case "broadcast":
+			conn.SetDeadline(time.Now().Add(sess.ExchangeTimeout))
+			err := proto.WriteMessage(protocol.PingPongMsg{Type: "ack"})
+			if err != nil {
+				return err
+			}
+			pack, err := json.Marshal(recv.Payloads)
+			if err != nil {
+				return err
+			}
+			events <- fmt.Sprintf("broadcast chan:%v app:%v topLevel:%d payloads:%s", recv.ChanId, recv.AppId, recv.TopLevel, pack)
+		}
+	}
+	return nil
+}

=== added directory 'server/acceptance/cmd'
=== added file 'server/acceptance/cmd/acceptanceclient.go'
--- server/acceptance/cmd/acceptanceclient.go	1970-01-01 00:00:00 +0000
+++ server/acceptance/cmd/acceptanceclient.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,79 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// acceptanceclient for playing
+package main
+
+import (
+	"launchpad.net/ubuntu-push/config"
+	"launchpad.net/ubuntu-push/server/acceptance"
+	"log"
+	"os"
+	"path/filepath"
+)
+
+type configuration struct {
+	// session configuration
+	PingInterval    config.ConfigTimeDuration `json:"ping_interval"`
+	ExchangeTimeout config.ConfigTimeDuration `json:"exchange_timeout"`
+	// server connection config
+	Addr        config.ConfigHostPort
+	CertPEMFile string `json:"cert_pem_file"`
+}
+
+func main() {
+	// xxx configure logging
+	switch {
+	case len(os.Args) < 2:
+		log.Fatal("missing config file")
+	case len(os.Args) < 3:
+		log.Fatal("missing device-id")
+	}
+	configFName := os.Args[1]
+	f, err := os.Open(configFName)
+	if err != nil {
+		log.Fatalf("reading config: %v", err)
+	}
+	cfg := &configuration{}
+	err = config.ReadConfig(f, cfg)
+	if err != nil {
+		log.Fatalf("reading config: %v", err)
+	}
+	session := &acceptance.ClientSession{
+		ExchangeTimeout: cfg.ExchangeTimeout.TimeDuration(),
+		PingInterval:    cfg.PingInterval.TimeDuration(),
+		ServerAddr:      cfg.Addr.HostPort(),
+		DeviceId:        os.Args[2],
+	}
+	session.CertPEMBlock, err = config.LoadFile(cfg.CertPEMFile, filepath.Dir(configFName))
+	if err != nil {
+		log.Fatalf("reading CertPEMFile: %v", err)
+	}
+	err = session.Dial()
+	if err != nil {
+		log.Fatalln(err)
+	}
+	events := make(chan string, 5)
+	go func() {
+		for {
+			log.Println(<-events)
+		}
+	}()
+	err = session.Run(events)
+	if err != nil {
+		log.Fatalln(err)
+	}
+}

=== added directory 'server/acceptance/config'
=== added file 'server/acceptance/config/README'
--- server/acceptance/config/README	1970-01-01 00:00:00 +0000
+++ server/acceptance/config/README	2014-01-14 15:09:46 +0000
@@ -0,0 +1,7 @@
+testing.key/testing.cert
+-------------------------
+generated with:
+
+go run /usr/lib/go/src/pkg/crypto/tls/generate_cert.go -ca -host localhost -rsa-bits 512 -duration 87600h
+
+and then renamed

=== added file 'server/acceptance/config/config.json'
--- server/acceptance/config/config.json	1970-01-01 00:00:00 +0000
+++ server/acceptance/config/config.json	2014-01-14 15:09:46 +0000
@@ -0,0 +1,12 @@
+{
+    "exchange_timeout": "0.5s",
+    "ping_interval": "2s",
+    "broker_queue_size": 100,
+    "session_queue_size": 10,
+    "addr": "127.0.0.1:9090",
+    "key_pem_file": "testing.key",
+    "cert_pem_file": "testing.cert",
+    "http_addr": "127.0.0.1:8888",
+    "http_read_timeout": "5s",
+    "http_write_timeout": "5s"
+}

=== added file 'server/acceptance/config/testing.cert'
--- server/acceptance/config/testing.cert	1970-01-01 00:00:00 +0000
+++ server/acceptance/config/testing.cert	2014-01-14 15:09:46 +0000
@@ -0,0 +1,10 @@
+-----BEGIN CERTIFICATE-----
+MIIBYzCCAQ+gAwIBAgIBADALBgkqhkiG9w0BAQUwEjEQMA4GA1UEChMHQWNtZSBD
+bzAeFw0xMzEyMTkyMDU1NDNaFw0yMzEyMTcyMDU1NDNaMBIxEDAOBgNVBAoTB0Fj
+bWUgQ28wWjALBgkqhkiG9w0BAQEDSwAwSAJBAPw+niki17X2qALE2A2AzE1q5dvK
+9CI4OduRtT9IgbFLC6psqAT21NA+QbY17nWSSpyP65zkMkwKXrbDzstwLPkCAwEA
+AaNUMFIwDgYDVR0PAQH/BAQDAgCkMBMGA1UdJQQMMAoGCCsGAQUFBwMBMA8GA1Ud
+EwEB/wQFMAMBAf8wGgYDVR0RBBMwEYIJbG9jYWxob3N0hwR/AAABMAsGCSqGSIb3
+DQEBBQNBAFqiVI+Km2XPSO+pxITaPvhmuzg+XG3l1+2di3gL+HlDobocjBqRctRU
+YySO32W07acjGJmCHUKpCJuq9X8hpmk=
+-----END CERTIFICATE-----

=== added file 'server/acceptance/config/testing.key'
--- server/acceptance/config/testing.key	1970-01-01 00:00:00 +0000
+++ server/acceptance/config/testing.key	2014-01-14 15:09:46 +0000
@@ -0,0 +1,9 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIBPAIBAAJBAPw+niki17X2qALE2A2AzE1q5dvK9CI4OduRtT9IgbFLC6psqAT2
+1NA+QbY17nWSSpyP65zkMkwKXrbDzstwLPkCAwEAAQJAKwXbIBULScP6QA6m8xam
+wgWbkvN41GVWqPafPV32kPBvKwSc+M1e+JR7g3/xPZE7TCELcfYi4yXEHZZI3Pbh
+oQIhAP/UsgJbsfH1GFv8Y8qGl5l/kmwwkwHhuKvEC87Yur9FAiEA/GlQv3ZfaXnT
+lcCFT0aL02O0RDiRYyMUG/JAZQJs6CUCIQCHO5SZYIUwxIGK5mCNxxXOAzyQSiD7
+hqkKywf+4FvfDQIhALa0TLyqJFom0t7c4iIGAIRc8UlIYQSPiajI64+x9775AiEA
+0v4fgSK/Rq059zW1753JjuB6aR0Uh+3RqJII4dUR1Wg=
+-----END RSA PRIVATE KEY-----

=== added directory 'server/api'
=== added file 'server/api/handlers.go'
--- server/api/handlers.go	1970-01-01 00:00:00 +0000
+++ server/api/handlers.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,244 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// Package api has code that offers a REST API for the applications that
+// want to push messages.
+package api
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/server/broker"
+	"launchpad.net/ubuntu-push/server/store"
+	"log"
+	"net/http"
+)
+
+const MaxRequestBodyBytes = 4 * 1024
+const JSONMediaType = "application/json"
+
+// APIError represents a API error (both internally and as JSON in a response).
+type APIError struct {
+	// http status code
+	StatusCode int `json:"-"`
+	// machine readable label
+	ErrorLabel string `json:"error"`
+	// human message
+	Message string `json:"message"`
+}
+
+// machine readable error labels
+const (
+	ioError        = "io-error"
+	invalidRequest = "invalid-request"
+	unknownChannel = "unknown channel"
+	unavailable    = "unavailable"
+	internalError  = "internal"
+)
+
+func (apiErr *APIError) Error() string {
+	return fmt.Sprintf("api %s: %s", apiErr.ErrorLabel, apiErr.Message)
+}
+
+// Well-known prebuilt API errors
+var (
+	ErrNoContentLengthProvided = &APIError{
+		http.StatusLengthRequired,
+		invalidRequest,
+		"A Content-Length must be provided",
+	}
+	ErrRequestBodyEmpty = &APIError{
+		http.StatusBadRequest,
+		invalidRequest,
+		"Request body empty",
+	}
+	ErrRequestBodyTooLarge = &APIError{
+		http.StatusRequestEntityTooLarge,
+		invalidRequest,
+		"Request body too large",
+	}
+	ErrWrongContentType = &APIError{
+		http.StatusUnsupportedMediaType,
+		invalidRequest,
+		"Wrong content type, should be application/json",
+	}
+	ErrWrongRequestMethod = &APIError{
+		http.StatusMethodNotAllowed,
+		invalidRequest,
+		"Wrong request method, should be POST",
+	}
+	ErrMalformedJSONObject = &APIError{
+		http.StatusBadRequest,
+		invalidRequest,
+		"Malformed JSON Object",
+	}
+	ErrCouldNotReadBody = &APIError{
+		http.StatusBadRequest,
+		ioError,
+		"Could not read request body",
+	}
+	ErrUnknownChannel = &APIError{
+		http.StatusBadRequest,
+		unknownChannel,
+		"Unknown channel",
+	}
+	ErrUnknown = &APIError{
+		http.StatusInternalServerError,
+		internalError,
+		"Unknown error",
+	}
+	ErrCouldNotStoreNotification = &APIError{
+		http.StatusServiceUnavailable,
+		unavailable,
+		"Could not store notification",
+	}
+)
+
+type Message struct {
+	Registration string          `json:"registration"`
+	CoalesceTag  string          `json:"coalesce_tag"`
+	Data         json.RawMessage `json:"data"`
+}
+
+// Broadcast request JSON object.
+type Broadcast struct {
+	Channel     string          `json:"channel"`
+	ExpireAfter uint8           `json:"expire_after"`
+	Data        json.RawMessage `json:"data"`
+}
+
+func respondError(writer http.ResponseWriter, apiErr *APIError) {
+	wireError, err := json.Marshal(apiErr)
+	if err != nil {
+		// xxx general 500 framework
+		log.Println("The provided string could not be marshaled into an error:", err)
+		http.Error(writer, "Internal Server Error", http.StatusInternalServerError)
+		return
+	}
+	writer.Header().Set("Content-type", JSONMediaType)
+	writer.WriteHeader(apiErr.StatusCode)
+	writer.Write(wireError)
+}
+
+func checkContentLength(request *http.Request) *APIError {
+	if request.ContentLength == -1 {
+		return ErrNoContentLengthProvided
+	}
+	if request.ContentLength == 0 {
+		return ErrRequestBodyEmpty
+	}
+	if request.ContentLength > MaxRequestBodyBytes {
+		return ErrRequestBodyTooLarge
+	}
+	return nil
+}
+
+func checkRequestAsPost(request *http.Request) *APIError {
+	if err := checkContentLength(request); err != nil {
+		return err
+	}
+	if request.Header.Get("Content-Type") != JSONMediaType {
+		return ErrWrongContentType
+	}
+	if request.Method != "POST" {
+		return ErrWrongRequestMethod
+	}
+	return nil
+}
+
+func readBody(writer http.ResponseWriter, request *http.Request) ([]byte, *APIError) {
+	if err := checkRequestAsPost(request); err != nil {
+		return nil, err
+	}
+
+	body := make([]byte, request.ContentLength)
+	_, err := io.ReadFull(request.Body, body)
+
+	if err != nil {
+		return nil, ErrCouldNotReadBody
+	}
+
+	return body, nil
+}
+
+// state holds the interfaces to delegate to serving requests
+type state struct {
+	store  store.PendingStore
+	broker broker.BrokerSending
+	logger logger.Logger
+}
+
+type BroadcastHandler state
+
+func (h *BroadcastHandler) doBroadcast(bcast *Broadcast) *APIError {
+	chanId, err := h.store.GetInternalChannelId(bcast.Channel)
+	if err != nil {
+		switch err {
+		case store.ErrUnknownChannel:
+			return ErrUnknownChannel
+		default:
+			return ErrUnknown
+		}
+	}
+	// xxx ignoring expiration for now
+	err = h.store.AppendToChannel(chanId, bcast.Data)
+	if err != nil {
+		// assume this for now
+		return ErrCouldNotStoreNotification
+	}
+
+	h.broker.Broadcast(chanId)
+	return nil
+}
+
+func (h *BroadcastHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	body, apiErr := readBody(writer, request)
+
+	if apiErr != nil {
+		respondError(writer, apiErr)
+		return
+	}
+
+	broadcast := &Broadcast{}
+	err := json.Unmarshal(body, broadcast)
+
+	if err != nil {
+		respondError(writer, ErrMalformedJSONObject)
+		return
+	}
+
+	apiErr = h.doBroadcast(broadcast)
+	if apiErr != nil {
+		respondError(writer, apiErr)
+		return
+	}
+
+	writer.Header().Set("Content-Type", "application/json")
+	fmt.Fprintf(writer, `{"ok":true}`)
+}
+
+// MakeHandlersMux makes a handler that dispatches for the various API endpoints.
+func MakeHandlersMux(store store.PendingStore, broker broker.BrokerSending, logger logger.Logger) http.Handler {
+	mux := http.NewServeMux()
+	mux.Handle("/broadcast", &BroadcastHandler{
+		store:  store,
+		broker: broker,
+		logger: logger,
+	})
+	return mux
+}

=== added file 'server/api/handlers_test.go'
--- server/api/handlers_test.go	1970-01-01 00:00:00 +0000
+++ server/api/handlers_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,345 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package api
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/ioutil"
+	. "launchpad.net/gocheck"
+	"launchpad.net/ubuntu-push/server/store"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestHandlers(t *testing.T) { TestingT(t) }
+
+type handlersSuite struct {
+	messageEndpoint string
+	json            string
+	client          *http.Client
+	c               *C
+}
+
+var _ = Suite(&handlersSuite{})
+
+func (s *handlersSuite) SetUpTest(c *C) {
+	s.client = &http.Client{}
+}
+
+func (s *handlersSuite) TestAPIError(c *C) {
+	var apiErr error = &APIError{400, invalidRequest, "Message"}
+	c.Check(apiErr.Error(), Equals, "api invalid-request: Message")
+	wire, err := json.Marshal(apiErr)
+	c.Assert(err, IsNil)
+	c.Check(string(wire), Equals, `{"error":"invalid-request","message":"Message"}`)
+}
+
+type checkBrokerSending struct {
+	store    store.PendingStore
+	chanId   store.InternalChannelId
+	err      error
+	top      int64
+	payloads []json.RawMessage
+}
+
+func (cbsend *checkBrokerSending) Broadcast(chanId store.InternalChannelId) {
+	top, payloads, err := cbsend.store.GetChannelSnapshot(chanId)
+	cbsend.err = err
+	cbsend.chanId = chanId
+	cbsend.top = top
+	cbsend.payloads = payloads
+}
+
+func (s *handlersSuite) TestDoBroadcast(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	bsend := &checkBrokerSending{store: sto}
+	bh := &BroadcastHandler{sto, bsend, nil}
+	payload := json.RawMessage(`{"a": 1}`)
+	apiErr := bh.doBroadcast(&Broadcast{
+		Channel: "system",
+		Data:    payload,
+	})
+	c.Check(apiErr, IsNil)
+	c.Check(bsend.err, IsNil)
+	c.Check(bsend.chanId, Equals, store.SystemInternalChannelId)
+	c.Check(bsend.top, Equals, int64(1))
+	c.Check(bsend.payloads, DeepEquals, []json.RawMessage{payload})
+}
+
+func (s *handlersSuite) TestDoBroadcastUnknownChannel(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	bh := &BroadcastHandler{sto, nil, nil}
+	apiErr := bh.doBroadcast(&Broadcast{
+		Channel: "unknown",
+		Data:    json.RawMessage(`{"a": 1}`),
+	})
+	c.Check(apiErr, Equals, ErrUnknownChannel)
+}
+
+type interceptInMemoryPendingStore struct {
+	*store.InMemoryPendingStore
+	intercept func(meth string, err error) error
+}
+
+func (isto *interceptInMemoryPendingStore) GetInternalChannelId(channel string) (store.InternalChannelId, error) {
+	chanId, err := isto.InMemoryPendingStore.GetInternalChannelId(channel)
+	return chanId, isto.intercept("GetInternalChannelId", err)
+}
+
+func (isto *interceptInMemoryPendingStore) AppendToChannel(chanId store.InternalChannelId, payload json.RawMessage) error {
+	err := isto.InMemoryPendingStore.AppendToChannel(chanId, payload)
+	return isto.intercept("AppendToChannel", err)
+}
+
+func (s *handlersSuite) TestDoBroadcastUnknownError(c *C) {
+	sto := &interceptInMemoryPendingStore{
+		store.NewInMemoryPendingStore(),
+		func(meth string, err error) error {
+			return errors.New("other")
+		},
+	}
+	bh := &BroadcastHandler{sto, nil, nil}
+	apiErr := bh.doBroadcast(&Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"a": 1}`),
+	})
+	c.Check(apiErr, Equals, ErrUnknown)
+}
+
+func (s *handlersSuite) TestDoBroadcastCouldNotStoreNotification(c *C) {
+	sto := &interceptInMemoryPendingStore{
+		store.NewInMemoryPendingStore(),
+		func(meth string, err error) error {
+			if meth == "AppendToChannel" {
+				return errors.New("fail")
+			}
+			return err
+		},
+	}
+	bh := &BroadcastHandler{sto, nil, nil}
+	apiErr := bh.doBroadcast(&Broadcast{
+		Channel: "system",
+		Data:    json.RawMessage(`{"a": 1}`),
+	})
+	c.Check(apiErr, Equals, ErrCouldNotStoreNotification)
+}
+
+func newPostRequest(path string, message interface{}, server *httptest.Server) *http.Request {
+	packedMessage, err := json.Marshal(message)
+	if err != nil {
+		panic(err)
+	}
+	reader := bytes.NewReader(packedMessage)
+
+	url := server.URL + path
+	request, _ := http.NewRequest("POST", url, reader)
+	request.ContentLength = int64(reader.Len())
+	request.Header.Set("Content-Type", "application/json")
+
+	return request
+}
+
+func getResponseBody(response *http.Response) ([]byte, error) {
+	defer response.Body.Close()
+	return ioutil.ReadAll(response.Body)
+}
+
+func checkError(c *C, response *http.Response, apiErr *APIError) {
+	c.Check(response.StatusCode, Equals, apiErr.StatusCode)
+	c.Check(response.Header.Get("Content-Type"), Equals, "application/json")
+	error := &APIError{StatusCode: response.StatusCode}
+	body, err := getResponseBody(response)
+	c.Assert(err, IsNil)
+	err = json.Unmarshal(body, error)
+	c.Assert(err, IsNil)
+	c.Check(error, DeepEquals, apiErr)
+}
+
+type testBrokerSending struct {
+	chanId chan store.InternalChannelId
+}
+
+func (bsend testBrokerSending) Broadcast(chanId store.InternalChannelId) {
+	bsend.chanId <- chanId
+}
+
+func (s *handlersSuite) TestRespondsToBasicSystemBroadcast(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	bsend := testBrokerSending{make(chan store.InternalChannelId, 1)}
+	testServer := httptest.NewServer(MakeHandlersMux(sto, bsend, nil))
+	defer testServer.Close()
+
+	payload := json.RawMessage(`{"foo":"bar"}`)
+
+	request := newPostRequest("/broadcast", &Broadcast{
+		Channel:     "system",
+		ExpireAfter: 60,
+		Data:        payload,
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+
+	c.Check(response.StatusCode, Equals, http.StatusOK)
+	c.Check(response.Header.Get("Content-Type"), Equals, "application/json")
+	body, err := getResponseBody(response)
+	c.Assert(err, IsNil)
+	dest := make(map[string]bool)
+	err = json.Unmarshal(body, &dest)
+	c.Assert(err, IsNil)
+	c.Check(dest, DeepEquals, map[string]bool{"ok": true})
+
+	top, _, err := sto.GetChannelSnapshot(store.SystemInternalChannelId)
+	c.Assert(err, IsNil)
+	c.Check(top, Equals, int64(1))
+	c.Check(<-bsend.chanId, Equals, store.SystemInternalChannelId)
+}
+
+func (s *handlersSuite) TestFromBroadcastError(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	testServer := httptest.NewServer(MakeHandlersMux(sto, nil, nil))
+	defer testServer.Close()
+
+	payload := json.RawMessage(`{"foo":"bar"}`)
+
+	request := newPostRequest("/broadcast", &Broadcast{
+		Channel:     "unkown",
+		ExpireAfter: 60,
+		Data:        payload,
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrUnknownChannel)
+}
+
+func (s *handlersSuite) TestCannotBroadcastMalformedData(c *C) {
+	testServer := httptest.NewServer(&BroadcastHandler{})
+	defer testServer.Close()
+
+	packedMessage := []byte("{some bogus-message: ")
+	reader := bytes.NewReader(packedMessage)
+
+	request, err := http.NewRequest("POST", testServer.URL, reader)
+	c.Assert(err, IsNil)
+	request.ContentLength = int64(len(packedMessage))
+	request.Header.Set("Content-Type", "application/json")
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrMalformedJSONObject)
+}
+
+func (s *handlersSuite) TestCannotBroadcastTooBigMessages(c *C) {
+	testServer := httptest.NewServer(&BroadcastHandler{})
+	defer testServer.Close()
+
+	bigString := strings.Repeat("a", MaxRequestBodyBytes)
+	dataString := fmt.Sprintf(`"%v"`, bigString)
+
+	request := newPostRequest("/", &Broadcast{
+		Channel:     "some-channel",
+		ExpireAfter: 60,
+		Data:        json.RawMessage([]byte(dataString)),
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrRequestBodyTooLarge)
+}
+
+func (s *handlersSuite) TestCannotBroadcastWithoutContentLength(c *C) {
+	testServer := httptest.NewServer(&BroadcastHandler{})
+	defer testServer.Close()
+
+	dataString := `{"foo":"bar"}`
+
+	request := newPostRequest("/", &Broadcast{
+		Channel:     "some-channel",
+		ExpireAfter: 60,
+		Data:        json.RawMessage([]byte(dataString)),
+	}, testServer)
+	request.ContentLength = -1
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrNoContentLengthProvided)
+}
+
+func (s *handlersSuite) TestCannotBroadcastEmptyMessages(c *C) {
+	testServer := httptest.NewServer(&BroadcastHandler{})
+	defer testServer.Close()
+
+	packedMessage := make([]byte, 0)
+	reader := bytes.NewReader(packedMessage)
+
+	request, err := http.NewRequest("POST", testServer.URL, reader)
+	c.Assert(err, IsNil)
+	request.ContentLength = int64(len(packedMessage))
+	request.Header.Set("Content-Type", "application/json")
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrRequestBodyEmpty)
+}
+
+func (s *handlersSuite) TestCannotBroadcastNonJSONMessages(c *C) {
+	testServer := httptest.NewServer(&BroadcastHandler{})
+	defer testServer.Close()
+
+	dataString := `{"foo":"bar"}`
+
+	request := newPostRequest("/", &Broadcast{
+		Channel:     "some-channel",
+		ExpireAfter: 60,
+		Data:        json.RawMessage([]byte(dataString)),
+	}, testServer)
+	request.Header.Set("Content-Type", "text/plain")
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrWrongContentType)
+}
+
+func (s *handlersSuite) TestCannotBroadcastNonPostMessages(c *C) {
+	testServer := httptest.NewServer(&BroadcastHandler{})
+	defer testServer.Close()
+
+	dataString := `{"foo":"bar"}`
+	packedMessage, err := json.Marshal(&Broadcast{
+		Channel:     "some-channel",
+		ExpireAfter: 60,
+		Data:        json.RawMessage([]byte(dataString)),
+	})
+	s.c.Assert(err, IsNil)
+	reader := bytes.NewReader(packedMessage)
+
+	request, err := http.NewRequest("GET", testServer.URL, reader)
+	c.Assert(err, IsNil)
+	request.ContentLength = int64(len(packedMessage))
+	request.Header.Set("Content-Type", "application/json")
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+
+	checkError(c, response, ErrWrongRequestMethod)
+}

=== added directory 'server/broker'
=== added file 'server/broker/broker.go'
--- server/broker/broker.go	1970-01-01 00:00:00 +0000
+++ server/broker/broker.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,71 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// Package broker handles session registrations and delivery of messages
+// through sessions.
+package broker
+
+import (
+	"fmt"
+	"launchpad.net/ubuntu-push/protocol"
+	"launchpad.net/ubuntu-push/server/store"
+)
+
+// Broker is responsible for registring sessions and delivering messages through them.
+type Broker interface {
+	// Register the session.
+	Register(*protocol.ConnectMsg) (BrokerSession, error)
+	// Unregister the session.
+	Unregister(BrokerSession)
+}
+
+// BrokerSending is the notification sending facet of the broker.
+type BrokerSending interface {
+	// Broadcast channel.
+	Broadcast(chanId store.InternalChannelId)
+}
+
+// Exchange guides the session through performing an exchange, typically delivery.
+type Exchange interface {
+	Prepare(BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)
+	Acked(BrokerSession) error
+}
+
+// BrokerSession holds broker session state.
+type BrokerSession interface {
+	// SessionChannel returns the session control channel
+	// on which the session gets exchanges to perform.
+	SessionChannel() <-chan Exchange
+	// DeviceId returns the device id string.
+	DeviceId() string
+}
+
+// Session aborted error.
+type ErrAbort struct {
+	Reason string
+}
+
+func (ea *ErrAbort) Error() string {
+	return fmt.Sprintf("session aborted (%s)", ea.Reason)
+}
+
+// BrokerConfig gives access to the typical broker configuration.
+type BrokerConfig interface {
+	// SessionQueueSize gives the session queue size.
+	SessionQueueSize() uint
+	// BrokerQueueSize gives the internal broker queue size.
+	BrokerQueueSize() uint
+}

=== added file 'server/broker/simple.go'
--- server/broker/simple.go	1970-01-01 00:00:00 +0000
+++ server/broker/simple.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,260 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package broker
+
+import (
+	"encoding/json"
+	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/protocol"
+	"launchpad.net/ubuntu-push/server/store"
+	// "log"
+	"sync"
+)
+
+// simpleBroker implements broker.Broker for everything in just one process.
+type SimpleBroker struct {
+	sto    store.PendingStore
+	logger logger.Logger
+	// running state
+	runMutex sync.Mutex
+	running  bool
+	stop     chan bool
+	stopped  chan bool
+	// sessions
+	sessionCh        chan *simpleBrokerSession
+	registry         map[string]*simpleBrokerSession
+	sessionQueueSize uint
+	// delivery
+	deliveryCh chan *delivery
+}
+
+// simpleBrokerSession represents a session in the broker.
+type simpleBrokerSession struct {
+	registered bool
+	deviceId   string
+	done       chan bool
+	exchanges  chan Exchange
+	levels     map[store.InternalChannelId]int64
+	// for exchanges
+	broadcastMsg protocol.BroadcastMsg
+	ackMsg       protocol.AckMsg
+}
+
+type deliveryKind int
+
+const (
+	broadcastDelivery deliveryKind = iota
+)
+
+// delivery holds all the information to request a delivery
+type delivery struct {
+	kind   deliveryKind
+	chanId store.InternalChannelId
+}
+
+func (sess *simpleBrokerSession) SessionChannel() <-chan Exchange {
+	return sess.exchanges
+}
+
+func (sess *simpleBrokerSession) DeviceId() string {
+	return sess.deviceId
+}
+
+// NewSimpleBroker makes a new SimpleBroker.
+func NewSimpleBroker(sto store.PendingStore, cfg BrokerConfig, logger logger.Logger) *SimpleBroker {
+	sessionCh := make(chan *simpleBrokerSession, cfg.BrokerQueueSize())
+	deliveryCh := make(chan *delivery, cfg.BrokerQueueSize())
+	registry := make(map[string]*simpleBrokerSession)
+	return &SimpleBroker{
+		logger:           logger,
+		sto:              sto,
+		stop:             make(chan bool),
+		stopped:          make(chan bool),
+		registry:         registry,
+		sessionCh:        sessionCh,
+		deliveryCh:       deliveryCh,
+		sessionQueueSize: cfg.SessionQueueSize(),
+	}
+}
+
+// Start starts the broker.
+func (b *SimpleBroker) Start() {
+	b.runMutex.Lock()
+	defer b.runMutex.Unlock()
+	if b.running {
+		return
+	}
+	b.running = true
+	go b.run()
+}
+
+// Stop stops the broker.
+func (b *SimpleBroker) Stop() {
+	b.runMutex.Lock()
+	defer b.runMutex.Unlock()
+	if !b.running {
+		return
+	}
+	b.stop <- true
+	<-b.stopped
+	b.running = false
+}
+
+func (b *SimpleBroker) feedPending(sess *simpleBrokerSession) error {
+	// find relevant channels, for now only system
+	channels := []store.InternalChannelId{store.SystemInternalChannelId}
+	for _, chanId := range channels {
+		topLevel, payloads, err := b.sto.GetChannelSnapshot(chanId)
+		if err != nil {
+			// next broadcast will try again
+			b.logger.Errorf("unsuccessful feed pending, get channel snapshot for %v: %v", chanId, err)
+			continue
+		}
+		clientLevel := sess.levels[chanId]
+		if clientLevel != topLevel {
+			broadcastExchg := &simpleBroadcastExchange{
+				chanId:               chanId,
+				topLevel:             topLevel,
+				notificationPayloads: payloads,
+			}
+			sess.exchanges <- broadcastExchg
+		}
+	}
+	return nil
+}
+
+// Register registers a session with the broker. It feeds the session
+// pending notifications as well.
+func (b *SimpleBroker) Register(connect *protocol.ConnectMsg) (BrokerSession, error) {
+	// xxx sanity check DeviceId
+	levels := map[store.InternalChannelId]int64{}
+	for hexId, v := range connect.Levels {
+		id, err := store.HexToInternalChannelId(hexId)
+		if err != nil {
+			return nil, &ErrAbort{err.Error()}
+		}
+		levels[id] = v
+	}
+	sess := &simpleBrokerSession{
+		deviceId:  connect.DeviceId,
+		done:      make(chan bool),
+		exchanges: make(chan Exchange, b.sessionQueueSize),
+		levels:    levels,
+	}
+	b.sessionCh <- sess
+	<-sess.done
+	err := b.feedPending(sess)
+	if err != nil {
+		return nil, err
+	}
+	return sess, nil
+}
+
+// Unregister unregisters a session with the broker. Doesn't wait.
+func (b *SimpleBroker) Unregister(s BrokerSession) {
+	sess := s.(*simpleBrokerSession)
+	b.sessionCh <- sess
+}
+
+// run runs the agent logic of the broker.
+func (b *SimpleBroker) run() {
+Loop:
+	for {
+		select {
+		case <-b.stop:
+			b.stopped <- true
+			break Loop
+		case sess := <-b.sessionCh:
+			if sess.registered { // unregister
+				// unregister only current
+				if b.registry[sess.deviceId] == sess {
+					delete(b.registry, sess.deviceId)
+				}
+			} else { // register
+				b.registry[sess.deviceId] = sess
+				sess.registered = true
+				sess.done <- true
+			}
+		case delivery := <-b.deliveryCh:
+			switch delivery.kind {
+			case broadcastDelivery:
+				topLevel, payloads, err := b.sto.GetChannelSnapshot(delivery.chanId)
+				if err != nil {
+					// next broadcast will try again
+					b.logger.Errorf("unsuccessful broadcast, get channel snapshot for %v: %v", delivery.chanId, err)
+					continue Loop
+				}
+				broadcastExchg := &simpleBroadcastExchange{
+					chanId:               delivery.chanId,
+					topLevel:             topLevel,
+					notificationPayloads: payloads,
+				}
+				for _, sess := range b.registry {
+					sess.exchanges <- broadcastExchg
+				}
+			}
+		}
+	}
+}
+
+// Broadcast requests the broadcast for a channel.
+func (b *SimpleBroker) Broadcast(chanId store.InternalChannelId) {
+	b.deliveryCh <- &delivery{
+		kind:   broadcastDelivery,
+		chanId: chanId,
+	}
+}
+
+// Exchanges
+
+type simpleBroadcastExchange struct {
+	chanId               store.InternalChannelId
+	topLevel             int64
+	notificationPayloads []json.RawMessage
+}
+
+func filterByLevel(clientLevel, topLevel int64, payloads []json.RawMessage) []json.RawMessage {
+	c := int64(len(payloads))
+	delta := topLevel - clientLevel
+	if delta < c {
+		return payloads[c-delta:]
+	} else {
+		return payloads
+	}
+}
+
+func (sbe *simpleBroadcastExchange) Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error) {
+	simpleSess := sess.(*simpleBrokerSession)
+	simpleSess.broadcastMsg.Type = "broadcast"
+	clientLevel := simpleSess.levels[sbe.chanId]
+	payloads := filterByLevel(clientLevel, sbe.topLevel, sbe.notificationPayloads)
+	// xxx need an AppId as well, later
+	simpleSess.broadcastMsg.ChanId = store.InternalChannelIdToHex(sbe.chanId)
+	simpleSess.broadcastMsg.TopLevel = sbe.topLevel
+	simpleSess.broadcastMsg.Payloads = payloads
+	return &simpleSess.broadcastMsg, &simpleSess.ackMsg, nil
+}
+
+func (sbe *simpleBroadcastExchange) Acked(sess BrokerSession) error {
+	simpleSess := sess.(*simpleBrokerSession)
+	if simpleSess.ackMsg.Type != "ack" {
+		return &ErrAbort{"expected ACK message"}
+	}
+	// update levels
+	simpleSess.levels[sbe.chanId] = sbe.topLevel
+	return nil
+}

=== added file 'server/broker/simple_test.go'
--- server/broker/simple_test.go	1970-01-01 00:00:00 +0000
+++ server/broker/simple_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,327 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package broker
+
+import (
+	"encoding/json"
+	"errors"
+	. "launchpad.net/gocheck"
+	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/protocol"
+	"launchpad.net/ubuntu-push/server/store"
+	helpers "launchpad.net/ubuntu-push/testing"
+	// "log"
+	"testing"
+	"time"
+)
+
+func TestSimple(t *testing.T) { TestingT(t) }
+
+type simpleSuite struct{}
+
+var _ = Suite(&simpleSuite{})
+
+type testBrokerConfig struct{}
+
+func (tbc *testBrokerConfig) SessionQueueSize() uint {
+	return 10
+}
+
+func (tbc *testBrokerConfig) BrokerQueueSize() uint {
+	return 5
+}
+
+func (s *simpleSuite) TestNew(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, nil)
+	c.Check(cap(b.sessionCh), Equals, 5)
+	c.Check(len(b.registry), Equals, 0)
+	c.Check(b.sto, Equals, sto)
+}
+
+func (s *simpleSuite) TestStartStop(c *C) {
+	b := NewSimpleBroker(nil, &testBrokerConfig{}, nil)
+	b.Start()
+	c.Check(b.running, Equals, true)
+	b.Start()
+	b.Stop()
+	c.Check(b.running, Equals, false)
+	b.Stop()
+}
+
+func (s *simpleSuite) TestRegistration(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, nil)
+	b.Start()
+	defer b.Stop()
+	sess, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1", Levels: map[string]int64{"0": 5}})
+	c.Assert(err, IsNil)
+	c.Assert(b.registry["dev-1"], Equals, sess)
+	c.Assert(sess.DeviceId(), Equals, "dev-1")
+	c.Check(sess.(*simpleBrokerSession).levels, DeepEquals, map[store.InternalChannelId]int64{
+		store.SystemInternalChannelId: 5,
+	})
+	b.Unregister(sess)
+	// just to make sure the unregister was processed
+	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: ""})
+	c.Assert(err, IsNil)
+	c.Check(b.registry["dev-1"], IsNil)
+}
+
+func (s *simpleSuite) TestRegistrationBrokenLevels(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, nil)
+	b.Start()
+	defer b.Stop()
+	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1", Levels: map[string]int64{"z": 5}})
+	c.Check(err, FitsTypeOf, &ErrAbort{})
+}
+
+func (s *simpleSuite) TestFeedPending(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	notification1 := json.RawMessage(`{"m": "M"}`)
+	sto.AppendToChannel(store.SystemInternalChannelId, notification1)
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, nil)
+	sess := &simpleBrokerSession{
+		exchanges: make(chan Exchange, 1),
+	}
+	b.feedPending(sess)
+	c.Assert(len(sess.exchanges), Equals, 1)
+	exchg1 := <-sess.exchanges
+	c.Check(exchg1, DeepEquals, &simpleBroadcastExchange{
+		chanId:               store.SystemInternalChannelId,
+		topLevel:             1,
+		notificationPayloads: []json.RawMessage{notification1},
+	})
+}
+
+func (s *simpleSuite) TestFeedPendingNop(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	notification1 := json.RawMessage(`{"m": "M"}`)
+	sto.AppendToChannel(store.SystemInternalChannelId, notification1)
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, nil)
+	sess := &simpleBrokerSession{
+		exchanges: make(chan Exchange, 1),
+		levels: map[store.InternalChannelId]int64{
+			store.SystemInternalChannelId: 1,
+		},
+	}
+	b.feedPending(sess)
+	c.Assert(len(sess.exchanges), Equals, 0)
+}
+
+func (s *simpleSuite) TestRegistrationFeedPending(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	notification1 := json.RawMessage(`{"m": "M"}`)
+	sto.AppendToChannel(store.SystemInternalChannelId, notification1)
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, nil)
+	b.Start()
+	defer b.Stop()
+	sess, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	c.Assert(err, IsNil)
+	c.Check(len(sess.(*simpleBrokerSession).exchanges), Equals, 1)
+}
+
+func (s *simpleSuite) TestRegistrationFeedPendingError(c *C) {
+	buf := &helpers.SyncedLogBuffer{}
+	logger := logger.NewSimpleLogger(buf, "error")
+	sto := &testFailingStore{}
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, logger)
+	b.Start()
+	defer b.Stop()
+	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	c.Assert(err, IsNil)
+	// but
+	c.Check(buf.String(), Matches, ".*ERROR unsuccessful feed pending, get channel snapshot for 0: get channel snapshot fail\n")
+}
+
+func (s *simpleSuite) TestRegistrationLastWins(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, nil)
+	b.Start()
+	defer b.Stop()
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	c.Assert(err, IsNil)
+	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	c.Assert(err, IsNil)
+	c.Assert(b.registry["dev-1"], Equals, sess2)
+	b.Unregister(sess1)
+	// just to make sure the unregister was processed
+	_, err = b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: ""})
+	c.Assert(err, IsNil)
+	c.Check(b.registry["dev-1"], Equals, sess2)
+}
+
+func (s *simpleSuite) TestBroadcastExchange(c *C) {
+	sess := &simpleBrokerSession{
+		levels: map[store.InternalChannelId]int64{},
+	}
+	exchg := &simpleBroadcastExchange{
+		chanId:   store.SystemInternalChannelId,
+		topLevel: 3,
+		notificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"a":"x"}`),
+			json.RawMessage(`{"a":"y"}`),
+		},
+	}
+	inMsg, outMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	// check
+	marshalled, err := json.Marshal(inMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"x"},{"a":"y"}]}`)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess)
+	c.Assert(err, IsNil)
+	c.Check(sess.levels[store.SystemInternalChannelId], Equals, int64(3))
+}
+
+func (s *simpleSuite) TestBroadcastExchangeAckMismatch(c *C) {
+	sess := &simpleBrokerSession{
+		levels: map[store.InternalChannelId]int64{},
+	}
+	exchg := &simpleBroadcastExchange{
+		chanId:   store.SystemInternalChannelId,
+		topLevel: 3,
+		notificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"a":"y"}`),
+		},
+	}
+	inMsg, outMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	// check
+	marshalled, err := json.Marshal(inMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
+	err = json.Unmarshal([]byte(`{}`), outMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess)
+	c.Assert(err, Not(IsNil))
+	c.Check(sess.levels[store.SystemInternalChannelId], Equals, int64(0))
+}
+
+func (s *simpleSuite) TestFilterByLevel(c *C) {
+	payloads := []json.RawMessage{
+		json.RawMessage(`{"a": 3}`),
+		json.RawMessage(`{"a": 4}`),
+		json.RawMessage(`{"a": 5}`),
+	}
+	res := filterByLevel(5, 5, payloads)
+	c.Check(len(res), Equals, 0)
+	res = filterByLevel(4, 5, payloads)
+	c.Check(len(res), Equals, 1)
+	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 5}`))
+	res = filterByLevel(3, 5, payloads)
+	c.Check(len(res), Equals, 2)
+	c.Check(res[0], DeepEquals, json.RawMessage(`{"a": 4}`))
+	res = filterByLevel(2, 5, payloads)
+	c.Check(len(res), Equals, 3)
+	res = filterByLevel(1, 5, payloads)
+	c.Check(len(res), Equals, 3)
+}
+
+func (s *simpleSuite) TestBroadcastExchangeFilterByLevel(c *C) {
+	sess := &simpleBrokerSession{
+		levels: map[store.InternalChannelId]int64{
+			store.SystemInternalChannelId: 2,
+		},
+	}
+	exchg := &simpleBroadcastExchange{
+		chanId:   store.SystemInternalChannelId,
+		topLevel: 3,
+		notificationPayloads: []json.RawMessage{
+			json.RawMessage(`{"a":"x"}`),
+			json.RawMessage(`{"a":"y"}`),
+		},
+	}
+	inMsg, outMsg, err := exchg.Prepare(sess)
+	c.Assert(err, IsNil)
+	// check
+	marshalled, err := json.Marshal(inMsg)
+	c.Assert(err, IsNil)
+	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
+	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
+	c.Assert(err, IsNil)
+	err = exchg.Acked(sess)
+	c.Assert(err, IsNil)
+}
+
+func (s *simpleSuite) TestBroadcast(c *C) {
+	sto := store.NewInMemoryPendingStore()
+	notification1 := json.RawMessage(`{"m": "M"}`)
+	sto.AppendToChannel(store.SystemInternalChannelId, notification1)
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, nil)
+	b.Start()
+	defer b.Stop()
+	sess1, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	c.Assert(err, IsNil)
+	sess2, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-2"})
+	c.Assert(err, IsNil)
+	b.Broadcast(store.SystemInternalChannelId)
+	select {
+	case <-time.After(5 * time.Second):
+		c.Fatal("taking too long to get broadcast exchange")
+	case exchg1 := <-sess1.SessionChannel():
+		c.Check(exchg1, DeepEquals, &simpleBroadcastExchange{
+			chanId:               store.SystemInternalChannelId,
+			topLevel:             1,
+			notificationPayloads: []json.RawMessage{notification1},
+		})
+	}
+	select {
+	case <-time.After(5 * time.Second):
+		c.Fatal("taking too long to get broadcast exchange")
+	case exchg2 := <-sess2.SessionChannel():
+		c.Check(exchg2, DeepEquals, &simpleBroadcastExchange{
+			chanId:               store.SystemInternalChannelId,
+			topLevel:             1,
+			notificationPayloads: []json.RawMessage{notification1},
+		})
+	}
+}
+
+type testFailingStore struct {
+	store.InMemoryPendingStore
+	countdownToFail int
+}
+
+func (sto *testFailingStore) GetChannelSnapshot(chanId store.InternalChannelId) (int64, []json.RawMessage, error) {
+	if sto.countdownToFail == 0 {
+		return 0, nil, errors.New("get channel snapshot fail")
+	}
+	sto.countdownToFail--
+	return 0, nil, nil
+}
+
+func (s *simpleSuite) TestBroadcastFail(c *C) {
+	buf := &helpers.SyncedLogBuffer{Written: make(chan bool, 1)}
+	logger := logger.NewSimpleLogger(buf, "error")
+	sto := &testFailingStore{countdownToFail: 1}
+	b := NewSimpleBroker(sto, &testBrokerConfig{}, logger)
+	b.Start()
+	defer b.Stop()
+	_, err := b.Register(&protocol.ConnectMsg{Type: "connect", DeviceId: "dev-1"})
+	c.Assert(err, IsNil)
+	b.Broadcast(store.SystemInternalChannelId)
+	select {
+	case <-time.After(5 * time.Second):
+		c.Fatal("taking too long to log error")
+	case <-buf.Written:
+	}
+	c.Check(buf.String(), Matches, ".*ERROR unsuccessful broadcast, get channel snapshot for 0: get channel snapshot fail\n")
+}

=== added directory 'server/dev'
=== added file 'server/dev/config.go'
--- server/dev/config.go	1970-01-01 00:00:00 +0000
+++ server/dev/config.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,110 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package main
+
+import (
+	"fmt"
+	"io"
+	"launchpad.net/ubuntu-push/config"
+	"time"
+)
+
+// expectedConfiguration is used as target for JSON parsing the configuration.
+type expectedConfiguration struct {
+	// session configuration
+	PingInterval    config.ConfigTimeDuration `json:"ping_interval"`
+	ExchangeTimeout config.ConfigTimeDuration `json:"exchange_timeout"`
+	// broker configuration
+	SessionQueueSize config.ConfigQueueSize `json:"session_queue_size"`
+	BrokerQueueSize  config.ConfigQueueSize `json:"broker_queue_size"`
+	// device listener configuration
+	Addr        config.ConfigHostPort `json:"addr"`
+	KeyPEMFile  string                `json:"key_pem_file"`
+	CertPEMFile string                `json:"cert_pem_file"`
+	// api http server configuration
+	HTTPAddr         config.ConfigHostPort     `json:"http_addr"`
+	HTTPReadTimeout  config.ConfigTimeDuration `json:"http_read_timeout"`
+	HTTPWriteTimeout config.ConfigTimeDuration `json:"http_write_timeout"`
+}
+
+// configuration holds the server configuration and gives it access
+// through the various component config interfaces.
+type configuration struct {
+	parsed       expectedConfiguration
+	certPEMBlock []byte
+	keyPEMBlock  []byte
+}
+
+func (cfg *configuration) PingInterval() time.Duration {
+	return cfg.parsed.PingInterval.TimeDuration()
+}
+
+func (cfg *configuration) ExchangeTimeout() time.Duration {
+	return cfg.parsed.ExchangeTimeout.TimeDuration()
+}
+
+func (cfg *configuration) SessionQueueSize() uint {
+	return cfg.parsed.SessionQueueSize.QueueSize()
+}
+
+func (cfg *configuration) BrokerQueueSize() uint {
+	return cfg.parsed.BrokerQueueSize.QueueSize()
+}
+
+func (cfg *configuration) Addr() string {
+	return cfg.parsed.Addr.HostPort()
+}
+
+func (cfg *configuration) KeyPEMBlock() []byte {
+	return cfg.keyPEMBlock
+}
+
+func (cfg *configuration) CertPEMBlock() []byte {
+	return cfg.certPEMBlock
+}
+
+func (cfg *configuration) HTTPAddr() string {
+	return cfg.parsed.HTTPAddr.HostPort()
+}
+
+func (cfg *configuration) HTTPReadTimeout() time.Duration {
+	return cfg.parsed.HTTPReadTimeout.TimeDuration()
+}
+
+func (cfg *configuration) HTTPWriteTimeout() time.Duration {
+	return cfg.parsed.HTTPWriteTimeout.TimeDuration()
+}
+
+// read reads & parses configuration from the reader. it uses baseDir
+// to load mentioned files in the configuration.
+func (cfg *configuration) read(r io.Reader, baseDir string) error {
+	err := config.ReadConfig(r, &cfg.parsed)
+	if err != nil {
+		return err
+	}
+	keyPEMBlock, err := config.LoadFile(cfg.parsed.KeyPEMFile, baseDir)
+	if err != nil {
+		return fmt.Errorf("reading key_pem_file: %v", err)
+	}
+	certPEMBlock, err := config.LoadFile(cfg.parsed.CertPEMFile, baseDir)
+	if err != nil {
+		return fmt.Errorf("reading cert_pem_file: %v", err)
+	}
+	cfg.keyPEMBlock = keyPEMBlock
+	cfg.certPEMBlock = certPEMBlock
+	return nil
+}

=== added file 'server/dev/http.go'
--- server/dev/http.go	1970-01-01 00:00:00 +0000
+++ server/dev/http.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,40 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package main
+
+import (
+	"net"
+	"net/http"
+	"time"
+)
+
+// A HTTPServeConfig holds the HTTP server config.
+type HTTPServeConfig interface {
+	HTTPAddr() string
+	HTTPReadTimeout() time.Duration
+	HTTPWriteTimeout() time.Duration
+}
+
+// RunHTTPServe serves HTTP requests.
+func RunHTTPServe(lst net.Listener, h http.Handler, cfg HTTPServeConfig) error {
+	srv := &http.Server{
+		Handler:      h,
+		ReadTimeout:  cfg.HTTPReadTimeout(),
+		WriteTimeout: cfg.HTTPReadTimeout(),
+	}
+	return srv.Serve(lst)
+}

=== added file 'server/dev/http_test.go'
--- server/dev/http_test.go	1970-01-01 00:00:00 +0000
+++ server/dev/http_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,70 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package main
+
+import (
+	"fmt"
+	"io/ioutil"
+	. "launchpad.net/gocheck"
+	// "log"
+	"net"
+	"net/http"
+	"time"
+)
+
+type httpSuite struct{}
+
+var _ = Suite(&httpSuite{})
+
+type testHTTPServeConfig struct{}
+
+func (cfg testHTTPServeConfig) HTTPAddr() string {
+	return "127.0.0.1:0"
+}
+
+func (cfg testHTTPServeConfig) HTTPReadTimeout() time.Duration {
+	return 5 * time.Second
+}
+
+func (cfg testHTTPServeConfig) HTTPWriteTimeout() time.Duration {
+	return 5 * time.Second
+}
+
+func testHandle(w http.ResponseWriter, r *http.Request) {
+	fmt.Fprintf(w, "yay!\n")
+}
+
+func (s *httpSuite) TestRunHTTPServe(c *C) {
+	cfg := testHTTPServeConfig{}
+	lst, err := net.Listen("tcp", cfg.HTTPAddr())
+	c.Assert(err, IsNil)
+	defer lst.Close()
+	errCh := make(chan error, 1)
+	h := http.HandlerFunc(testHandle)
+	go func() {
+		errCh <- RunHTTPServe(lst, h, cfg)
+	}()
+	resp, err := http.Get(fmt.Sprintf("http://%s/", lst.Addr()))
+	c.Assert(err, IsNil)
+	defer resp.Body.Close()
+	c.Assert(resp.StatusCode, Equals, 200)
+	body, err := ioutil.ReadAll(resp.Body)
+	c.Assert(err, IsNil)
+	c.Check(string(body), Equals, "yay!\n")
+	lst.Close()
+	c.Check(<-errCh, ErrorMatches, ".*closed.*")
+}

=== added file 'server/dev/server.go'
--- server/dev/server.go	1970-01-01 00:00:00 +0000
+++ server/dev/server.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,77 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// simple development server.
+package main
+
+import (
+	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/server/api"
+	"launchpad.net/ubuntu-push/server/broker"
+	"launchpad.net/ubuntu-push/server/listener"
+	"launchpad.net/ubuntu-push/server/session"
+	"launchpad.net/ubuntu-push/server/store"
+	"net"
+	"os"
+	"path/filepath"
+)
+
+func main() {
+	logger := logger.NewSimpleLogger(os.Stderr, "debug")
+	if len(os.Args) < 2 { // xxx use flag
+		logger.Fatalf("missing config file")
+	}
+	configFName := os.Args[1]
+	f, err := os.Open(configFName)
+	if err != nil {
+		logger.Fatalf("reading config: %v", err)
+	}
+	cfg := &configuration{}
+	err = cfg.read(f, filepath.Dir(configFName))
+	if err != nil {
+		logger.Fatalf("reading config: %v", err)
+	}
+	// setup a pending store and start the broker
+	sto := store.NewInMemoryPendingStore()
+	broker := broker.NewSimpleBroker(sto, cfg, logger)
+	broker.Start()
+	defer broker.Stop()
+	// serve the http api
+	httpLst, err := net.Listen("tcp", cfg.HTTPAddr())
+	if err != nil {
+		logger.Fatalf("start http listening: %v", err)
+	}
+	handler := api.MakeHandlersMux(sto, broker, logger)
+	logger.Infof("listening for http on %v", httpLst.Addr())
+	go func() {
+		err := RunHTTPServe(httpLst, handler, cfg)
+		if err != nil {
+			logger.Fatalf("accepting http connections: %v", err)
+		}
+	}()
+	// listen for device connections
+	lst, err := listener.DeviceListen(cfg)
+	if err != nil {
+		logger.Fatalf("start device listening: %v", err)
+	}
+	logger.Infof("listening for devices on %v", lst.Addr())
+	err = lst.AcceptLoop(func(conn net.Conn) error {
+		return session.Session(conn, broker, cfg, logger)
+	}, logger)
+	if err != nil {
+		logger.Fatalf("accepting device connections: %v", err)
+	}
+}

=== added file 'server/dev/server_test.go'
--- server/dev/server_test.go	1970-01-01 00:00:00 +0000
+++ server/dev/server_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,105 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package main
+
+import (
+	// "fmt"
+	"bytes"
+	"io/ioutil"
+	. "launchpad.net/gocheck"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestDevserver(t *testing.T) { TestingT(t) }
+
+type devserverSuite struct{}
+
+var _ = Suite(&devserverSuite{})
+
+func (s *devserverSuite) TestConfigRead(c *C) {
+	tmpDir := c.MkDir()
+	err := ioutil.WriteFile(filepath.Join(tmpDir, "key.key"), []byte("KeY"), os.ModePerm)
+	c.Assert(err, IsNil)
+	err = ioutil.WriteFile(filepath.Join(tmpDir, "cert.cert"), []byte("CeRt"), os.ModePerm)
+	c.Assert(err, IsNil)
+	buf := bytes.NewBufferString(`{
+"ping_interval": "5m",
+"exchange_timeout": "10s",
+"session_queue_size": 10,
+"broker_queue_size": 100,
+"addr": "127.0.0.1:9999",
+"key_pem_file": "key.key",
+"cert_pem_file": "cert.cert",
+"http_addr": "127.0.0.1:8080",
+"http_read_timeout": "5s",
+"http_write_timeout": "10s"
+}`)
+	cfg := &configuration{}
+	err = cfg.read(buf, tmpDir)
+	c.Assert(err, IsNil)
+	c.Check(cfg.PingInterval(), Equals, 5*time.Minute)
+	c.Check(cfg.ExchangeTimeout(), Equals, 10*time.Second)
+	c.Check(cfg.BrokerQueueSize(), Equals, uint(100))
+	c.Check(cfg.SessionQueueSize(), Equals, uint(10))
+	c.Check(cfg.Addr(), Equals, "127.0.0.1:9999")
+	c.Check(string(cfg.KeyPEMBlock()), Equals, "KeY")
+	c.Check(string(cfg.CertPEMBlock()), Equals, "CeRt")
+	c.Check(cfg.HTTPAddr(), Equals, "127.0.0.1:8080")
+	c.Check(cfg.HTTPReadTimeout(), Equals, 5*time.Second)
+	c.Check(cfg.HTTPWriteTimeout(), Equals, 10*time.Second)
+}
+
+func (s *devserverSuite) TestConfigReadErrors(c *C) {
+	tmpDir := c.MkDir()
+	checkError := func(config, expectedErr string) {
+		cfg := &configuration{}
+		err := cfg.read(bytes.NewBufferString(config), tmpDir)
+		c.Check(err, ErrorMatches, expectedErr)
+	}
+	checkError("", "EOF")
+	checkError(`{"ping_interval": "1m"}`, "missing exchange_timeout")
+	checkError(`{"ping_interval": "1m", "exchange_timeout": "5s", "session_queue_size": "foo"}`, "session_queue_size:.*type uint")
+	checkError(`{
+"exchange_timeout": "10s",
+"ping_interval": "5m",
+"broker_queue_size": 100,
+"session_queue_size": 10,
+"addr": ":9000",
+"key_pem_file": "doesntexist",
+"cert_pem_file": "doesntexist",
+"http_addr": ":8080",
+"http_read_timeout": "5s",
+"http_write_timeout": "10s"
+}`, "reading key_pem_file:.*no such file.*")
+	err := ioutil.WriteFile(filepath.Join(tmpDir, "key.key"), []byte("KeY"), os.ModePerm)
+	c.Assert(err, IsNil)
+	checkError(`{
+"exchange_timeout": "10s",
+"ping_interval": "5m",
+"broker_queue_size": 100,
+"session_queue_size": 10,
+"addr": ":9000",
+"key_pem_file": "key.key",
+"cert_pem_file": "doesntexist",
+"http_addr": ":8080",
+"http_read_timeout": "5s",
+"http_write_timeout": "10s"
+}`, "reading cert_pem_file:.*no such file.*")
+}

=== added directory 'server/listener'
=== added file 'server/listener/listener.go'
--- server/listener/listener.go	1970-01-01 00:00:00 +0000
+++ server/listener/listener.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,86 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// Package listener has code to listen for device connections and setup sessions for them.
+package listener
+
+import (
+	"crypto/tls"
+	"launchpad.net/ubuntu-push/logger"
+	"net"
+	"time"
+)
+
+// A DeviceListenerConfig offers the DeviceListener configuration.
+type DeviceListenerConfig interface {
+	// Addr to listen on.
+	Addr() string
+	// TLS key
+	KeyPEMBlock() []byte
+	// TLS cert
+	CertPEMBlock() []byte
+}
+
+// DeviceListener listens and setup sessions from device connections.
+type DeviceListener struct {
+	net.Listener
+}
+
+// DeviceListen creates a DeviceListener for device connections based on config.
+func DeviceListen(cfg DeviceListenerConfig) (*DeviceListener, error) {
+	cert, err := tls.X509KeyPair(cfg.CertPEMBlock(), cfg.KeyPEMBlock())
+	if err != nil {
+		return nil, err
+	}
+	tlsCfg := &tls.Config{
+		Certificates:           []tls.Certificate{cert},
+		SessionTicketsDisabled: true,
+	}
+	lst, err := tls.Listen("tcp", cfg.Addr(), tlsCfg)
+	return &DeviceListener{lst}, err
+}
+
+// handleTemporary checks and handles if the error is just a temporary network
+// error.
+func handleTemporary(err error) bool {
+	if netError, isNetError := err.(net.Error); isNetError {
+		if netError.Temporary() {
+			// wait, xxx exponential backoff?
+			time.Sleep(100 * time.Millisecond)
+			return true
+		}
+	}
+	return false
+}
+
+// AcceptLoop accepts connections and starts sessions for them.
+func (dl *DeviceListener) AcceptLoop(session func(net.Conn) error, logger logger.Logger) error {
+	for {
+		// xxx enforce a connection limit
+		conn, err := dl.Listener.Accept()
+		if err != nil {
+			if handleTemporary(err) {
+				logger.Errorf("device listener: %s -- retrying", err)
+				continue
+			}
+			return err
+		}
+		go func() {
+			defer logger.Recoverf("terminating device connection")
+			session(conn)
+		}()
+	}
+}

=== added file 'server/listener/listener_test.go'
--- server/listener/listener_test.go	1970-01-01 00:00:00 +0000
+++ server/listener/listener_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,239 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package listener
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	. "launchpad.net/gocheck"
+	"launchpad.net/ubuntu-push/logger"
+	helpers "launchpad.net/ubuntu-push/testing"
+	"net"
+	"syscall"
+	"testing"
+	"time"
+)
+
+func TestListener(t *testing.T) { TestingT(t) }
+
+type listenerSuite struct{}
+
+var _ = Suite(&listenerSuite{})
+
+const NofileMax = 500
+
+func (s *listenerSuite) SetUpSuite(*C) {
+	// make it easier to get a too many open files error
+	var nofileLimit syscall.Rlimit
+	err := syscall.Getrlimit(syscall.RLIMIT_NOFILE, &nofileLimit)
+	if err != nil {
+		panic(err)
+	}
+	nofileLimit.Cur = NofileMax
+	err = syscall.Setrlimit(syscall.RLIMIT_NOFILE, &nofileLimit)
+	if err != nil {
+		panic(err)
+	}
+}
+
+type testDevListenerCfg struct {
+	addr string
+}
+
+func (cfg *testDevListenerCfg) Addr() string {
+	return cfg.addr
+}
+
+// key&cert generated with go run /usr/lib/go/src/pkg/crypto/tls/generate_cert.go -ca -host localhost -rsa-bits 512 -duration 87600h
+
+func (cfg *testDevListenerCfg) KeyPEMBlock() []byte {
+	return []byte(`-----BEGIN RSA PRIVATE KEY-----
+MIIBPAIBAAJBAPw+niki17X2qALE2A2AzE1q5dvK9CI4OduRtT9IgbFLC6psqAT2
+1NA+QbY17nWSSpyP65zkMkwKXrbDzstwLPkCAwEAAQJAKwXbIBULScP6QA6m8xam
+wgWbkvN41GVWqPafPV32kPBvKwSc+M1e+JR7g3/xPZE7TCELcfYi4yXEHZZI3Pbh
+oQIhAP/UsgJbsfH1GFv8Y8qGl5l/kmwwkwHhuKvEC87Yur9FAiEA/GlQv3ZfaXnT
+lcCFT0aL02O0RDiRYyMUG/JAZQJs6CUCIQCHO5SZYIUwxIGK5mCNxxXOAzyQSiD7
+hqkKywf+4FvfDQIhALa0TLyqJFom0t7c4iIGAIRc8UlIYQSPiajI64+x9775AiEA
+0v4fgSK/Rq059zW1753JjuB6aR0Uh+3RqJII4dUR1Wg=
+-----END RSA PRIVATE KEY-----`)
+}
+
+func (cfg *testDevListenerCfg) CertPEMBlock() []byte {
+	return []byte(`-----BEGIN CERTIFICATE-----
+MIIBYzCCAQ+gAwIBAgIBADALBgkqhkiG9w0BAQUwEjEQMA4GA1UEChMHQWNtZSBD
+bzAeFw0xMzEyMTkyMDU1NDNaFw0yMzEyMTcyMDU1NDNaMBIxEDAOBgNVBAoTB0Fj
+bWUgQ28wWjALBgkqhkiG9w0BAQEDSwAwSAJBAPw+niki17X2qALE2A2AzE1q5dvK
+9CI4OduRtT9IgbFLC6psqAT21NA+QbY17nWSSpyP65zkMkwKXrbDzstwLPkCAwEA
+AaNUMFIwDgYDVR0PAQH/BAQDAgCkMBMGA1UdJQQMMAoGCCsGAQUFBwMBMA8GA1Ud
+EwEB/wQFMAMBAf8wGgYDVR0RBBMwEYIJbG9jYWxob3N0hwR/AAABMAsGCSqGSIb3
+DQEBBQNBAFqiVI+Km2XPSO+pxITaPvhmuzg+XG3l1+2di3gL+HlDobocjBqRctRU
+YySO32W07acjGJmCHUKpCJuq9X8hpmk=
+-----END CERTIFICATE-----`)
+}
+
+func (s *listenerSuite) TestDeviceListen(c *C) {
+	lst, err := DeviceListen(&testDevListenerCfg{"127.0.0.1:0"})
+	c.Check(err, IsNil)
+	defer lst.Close()
+	c.Check(lst.Addr().String(), Matches, `127.0.0.1:\d{5}`)
+}
+
+func (s *listenerSuite) TestDeviceListenError(c *C) {
+	// assume tests are not running as root
+	_, err := DeviceListen(&testDevListenerCfg{"127.0.0.1:99"})
+	c.Check(err, ErrorMatches, ".*permission denied.*")
+}
+
+type testNetError struct {
+	temp bool
+}
+
+func (tne *testNetError) Error() string {
+	return "test net error"
+}
+
+func (tne *testNetError) Temporary() bool {
+	return tne.temp
+}
+
+func (tne *testNetError) Timeout() bool {
+	return false
+}
+
+var _ net.Error = &testNetError{} // sanity check
+
+func (s *listenerSuite) TestHandleTemporary(c *C) {
+	c.Check(handleTemporary(&testNetError{true}), Equals, true)
+	c.Check(handleTemporary(&testNetError{false}), Equals, false)
+}
+
+func testSession(conn net.Conn) error {
+	defer conn.Close()
+	conn.SetDeadline(time.Now().Add(2 * time.Second))
+	var buf [1]byte
+	_, err := conn.Read(buf[:])
+	if err != nil {
+		return err
+	}
+	_, err = conn.Write(buf[:])
+	return err
+}
+
+func testTlsDial(c *C, addr string) (net.Conn, error) {
+	cp := x509.NewCertPool()
+	ok := cp.AppendCertsFromPEM((&testDevListenerCfg{}).CertPEMBlock())
+	c.Assert(ok, Equals, true)
+	return tls.Dial("tcp", addr, &tls.Config{RootCAs: cp})
+}
+
+func testWriteByte(c *C, conn net.Conn, toWrite uint32) {
+	conn.SetDeadline(time.Now().Add(2 * time.Second))
+	_, err := conn.Write([]byte{byte(toWrite)})
+	c.Assert(err, IsNil)
+}
+
+func testReadByte(c *C, conn net.Conn, expected uint32) {
+	var buf [1]byte
+	_, err := conn.Read(buf[:])
+	c.Check(err, IsNil)
+	c.Check(buf[0], Equals, byte(expected))
+}
+
+func (s *listenerSuite) TestDeviceAcceptLoop(c *C) {
+	buf := &helpers.SyncedLogBuffer{}
+	logger := logger.NewSimpleLogger(buf, "error")
+	lst, err := DeviceListen(&testDevListenerCfg{"127.0.0.1:0"})
+	c.Check(err, IsNil)
+	defer lst.Close()
+	errCh := make(chan error)
+	go func() {
+		errCh <- lst.AcceptLoop(testSession, logger)
+	}()
+	listenerAddr := lst.Addr().String()
+	conn1, err := testTlsDial(c, listenerAddr)
+	c.Assert(err, IsNil)
+	defer conn1.Close()
+	testWriteByte(c, conn1, '1')
+	conn2, err := testTlsDial(c, listenerAddr)
+	c.Assert(err, IsNil)
+	defer conn2.Close()
+	testWriteByte(c, conn2, '2')
+	testReadByte(c, conn1, '1')
+	testReadByte(c, conn2, '2')
+	lst.Close()
+	c.Check(<-errCh, ErrorMatches, ".*use of closed.*")
+	c.Check(buf.String(), Equals, "")
+}
+
+func (s *listenerSuite) TestDeviceAcceptLoopTemporaryError(c *C) {
+	buf := &helpers.SyncedLogBuffer{}
+	logger := logger.NewSimpleLogger(buf, "error")
+	// ENFILE is not the temp network error we want to handle this way
+	// but is relatively easy to generate in a controlled way
+	var err error
+	lst, err := DeviceListen(&testDevListenerCfg{"127.0.0.1:0"})
+	c.Check(err, IsNil)
+	defer lst.Close()
+	errCh := make(chan error)
+	go func() {
+		errCh <- lst.AcceptLoop(testSession, logger)
+	}()
+	listenerAddr := lst.Addr().String()
+	conns := make([]net.Conn, 0, NofileMax)
+	for i := 0; i < NofileMax; i++ {
+		var conn1 net.Conn
+		conn1, err = net.Dial("tcp", listenerAddr)
+		if err != nil {
+			break
+		}
+		defer conn1.Close()
+		conns = append(conns, conn1)
+	}
+	c.Assert(err, ErrorMatches, "*.too many open.*")
+	for _, conn := range conns {
+		conn.Close()
+	}
+	conn2, err := testTlsDial(c, listenerAddr)
+	c.Assert(err, IsNil)
+	defer conn2.Close()
+	testWriteByte(c, conn2, '2')
+	testReadByte(c, conn2, '2')
+	lst.Close()
+	c.Check(<-errCh, ErrorMatches, ".*use of closed.*")
+	c.Check(buf.String(), Matches, ".*device listener:.*accept.*too many open.*-- retrying\n")
+}
+
+func (s *listenerSuite) TestDeviceAcceptLoopPanic(c *C) {
+	buf := &helpers.SyncedLogBuffer{}
+	logger := logger.NewSimpleLogger(buf, "error")
+	lst, err := DeviceListen(&testDevListenerCfg{"127.0.0.1:0"})
+	c.Check(err, IsNil)
+	defer lst.Close()
+	errCh := make(chan error)
+	go func() {
+		errCh <- lst.AcceptLoop(func(conn net.Conn) error {
+			defer conn.Close()
+			panic("session panic")
+		}, logger)
+	}()
+	listenerAddr := lst.Addr().String()
+	_, err = testTlsDial(c, listenerAddr)
+	c.Assert(err, Not(IsNil))
+	lst.Close()
+	c.Check(<-errCh, ErrorMatches, ".*use of closed.*")
+	c.Check(buf.String(), Matches, "(?s).*session panic!! terminating device connection:\n.*AcceptLoop.*")
+}

=== added directory 'server/session'
=== added file 'server/session/session.go'
--- server/session/session.go	1970-01-01 00:00:00 +0000
+++ server/session/session.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,157 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+// Package session has code handling long-lived connections from devices.
+package session
+
+import (
+	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/protocol"
+	"launchpad.net/ubuntu-push/server/broker"
+	"net"
+	"time"
+)
+
+// SessionConfig is for carrying the session configuration.
+type SessionConfig interface {
+	// pings are emitted each ping interval
+	PingInterval() time.Duration
+	// send and waiting for response shouldn't take more than exchange
+	// timeout
+	ExchangeTimeout() time.Duration
+}
+
+// sessionStart manages the start of the protocol session.
+func sessionStart(proto protocol.Protocol, brkr broker.Broker, cfg SessionConfig) (broker.BrokerSession, error) {
+	var connMsg protocol.ConnectMsg
+	proto.SetDeadline(time.Now().Add(cfg.ExchangeTimeout()))
+	err := proto.ReadMessage(&connMsg)
+	if err != nil {
+		return nil, err
+	}
+	if connMsg.Type != "connect" {
+		return nil, &broker.ErrAbort{"expected CONNECT message"}
+	}
+	return brkr.Register(&connMsg)
+}
+
+// exchange writes outMsg message, reads answer in inMsg
+func exchange(proto protocol.Protocol, outMsg, inMsg interface{}, exchangeTimeout time.Duration) error {
+	proto.SetDeadline(time.Now().Add(exchangeTimeout))
+	err := proto.WriteMessage(outMsg)
+	if err != nil {
+		return err
+	}
+	err = proto.ReadMessage(inMsg)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// sessionLoop manages the exchanges of the protocol session.
+func sessionLoop(proto protocol.Protocol, sess broker.BrokerSession, cfg SessionConfig) error {
+	pingInterval := cfg.PingInterval()
+	exchangeTimeout := cfg.ExchangeTimeout()
+	pingTimer := time.NewTimer(pingInterval)
+	ch := sess.SessionChannel()
+	for {
+		select {
+		case <-pingTimer.C:
+			pingMsg := &protocol.PingPongMsg{"ping"}
+			var pongMsg protocol.PingPongMsg
+			err := exchange(proto, pingMsg, &pongMsg, exchangeTimeout)
+			if err != nil {
+				return err
+			}
+			if pongMsg.Type != "pong" {
+				return &broker.ErrAbort{"expected PONG message"}
+			}
+			pingTimer.Reset(pingInterval)
+		case exchg := <-ch:
+			// xxx later can use ch closing for shutdown/reset
+			pingTimer.Stop()
+			outMsg, inMsg, err := exchg.Prepare(sess)
+			if err != nil {
+				return err
+			}
+			for {
+				done := outMsg.Split()
+				err = exchange(proto, outMsg, inMsg, exchangeTimeout)
+				if err != nil {
+					return err
+				}
+				if done {
+					pingTimer.Reset(pingInterval)
+				}
+				err = exchg.Acked(sess)
+				if err != nil {
+					return err
+				}
+				if done {
+					break
+				}
+			}
+		}
+	}
+	return nil
+}
+
+var sessionsEpoch = time.Date(2013, 1, 1, 0, 0, 0, 0, time.UTC).UnixNano()
+
+// sessionTracker logs session events.
+type sessionTracker struct {
+	logger.Logger
+	sessionId int64 // xxx use timeuuid later
+}
+
+func (trk *sessionTracker) start(conn interface {
+	RemoteAddr() net.Addr
+}) {
+	trk.sessionId = time.Now().UnixNano() - sessionsEpoch
+	trk.Debugf("session(%x) connected %v", trk.sessionId, conn.RemoteAddr())
+}
+
+func (trk *sessionTracker) registered(sess broker.BrokerSession) {
+	trk.Infof("session(%x) registered %v", trk.sessionId, sess.DeviceId())
+}
+
+func (trk *sessionTracker) end(err error) error {
+	trk.Debugf("session(%x) ended with: %v", trk.sessionId, err)
+	return err
+}
+
+// Session manages the session with a client.
+func Session(conn net.Conn, brkr broker.Broker, cfg SessionConfig, logger logger.Logger) error {
+	defer conn.Close()
+	track := sessionTracker{Logger: logger}
+	track.start(conn)
+	v, err := protocol.ReadWireFormatVersion(conn, cfg.ExchangeTimeout())
+	if err != nil {
+		return track.end(err)
+	}
+	if v != protocol.ProtocolWireVersion {
+		return track.end(&broker.ErrAbort{"unexpected wire format version"})
+	}
+	proto := protocol.NewProtocol0(conn)
+	sess, err := sessionStart(proto, brkr, cfg)
+	if err != nil {
+		return track.end(err)
+	}
+	track.registered(sess)
+	defer brkr.Unregister(sess)
+	return track.end(sessionLoop(proto, sess, cfg))
+}

=== added file 'server/session/session_test.go'
--- server/session/session_test.go	1970-01-01 00:00:00 +0000
+++ server/session/session_test.go	2014-01-14 15:09:46 +0000
@@ -0,0 +1,619 @@
+/*
+ Copyright 2013-2014 Canonical Ltd.
+
+ This program is free software: you can redistribute it and/or modify it
+ under the terms of the GNU General Public License version 3, as published
+ by the Free Software Foundation.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranties of
+ MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
+ PURPOSE.  See the GNU General Public License for more details.
+
+ You should have received a copy of the GNU General Public License along
+ with this program.  If not, see <http://www.gnu.org/licenses/>.
+*/
+
+package session
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"io/ioutil"
+	. "launchpad.net/gocheck"
+	"launchpad.net/ubuntu-push/logger"
+	"launchpad.net/ubuntu-push/protocol"
+	"launchpad.net/ubuntu-push/server/broker"
+	"net"
+	"reflect"
+	"testing"
+	"time"
+)
+
+func TestSession(t *testing.T) { TestingT(t) }
+
+type sessionSuite struct{}
+
+var _ = Suite(&sessionSuite{})
+
+type testProtocol struct {
+	up   chan interface{}
+	down chan interface{}
+}
+
+// takeNext takes a value from given channel with a 5s timeout
+func takeNext(ch <-chan interface{}) interface{} {
+	select {
+	case <-time.After(5 * time.Second):
+		panic("test protocol exchange stuck: too long waiting")
+	case v := <-ch:
+		return v
+	}
+	return nil
+}
+
+func (c *testProtocol) SetDeadline(t time.Time) {
+	deadAfter := t.Sub(time.Now())
+	deadAfter = (deadAfter + time.Millisecond/2) / time.Millisecond * time.Millisecond
+	c.down <- fmt.Sprintf("deadline %v", deadAfter)
+}
+
+func (c *testProtocol) ReadMessage(dest interface{}) error {
+	switch v := takeNext(c.up).(type) {
+	case error:
+		return v
+	default:
+		// make sure JSON.Unmarshal works with dest
+		var marshalledMsg []byte
+		marshalledMsg, err := json.Marshal(v)
+		if err != nil {
+			return fmt.Errorf("can't jsonify test value: %v", v)
+		}
+		return json.Unmarshal(marshalledMsg, dest)
+	}
+	return nil
+}
+
+func (c *testProtocol) WriteMessage(src interface{}) error {
+	// make sure JSON.Marshal works with src
+	_, err := json.Marshal(src)
+	if err != nil {
+		return err
+	}
+	val := reflect.ValueOf(src)
+	if val.Kind() == reflect.Ptr {
+		src = val.Elem().Interface()
+	}
+	c.down <- src
+	switch v := takeNext(c.up).(type) {
+	case error:
+		return v
+	}
+	return nil
+}
+
+type testSessionConfig struct {
+	exchangeTimeout time.Duration
+	pingInterval    time.Duration
+}
+
+func (tsc *testSessionConfig) PingInterval() time.Duration {
+	return tsc.pingInterval
+}
+
+func (tsc *testSessionConfig) ExchangeTimeout() time.Duration {
+	return tsc.exchangeTimeout
+}
+