wip

.gitignore

@@ -1,3 +1,4 @@
+
 bower_components/
 node_modules/
 .pulp-cache/
@@ -9,4 +10,3 @@ generated-docs/
 .purs*
 .psa*
 .spago
-.tmp/

LICENSE

@@ -1,675 +0,0 @@
-                    GNU GENERAL PUBLIC LICENSE
-                       Version 3, 29 June 2007
-
- Copyright (C) 2007 Free Software Foundation, Inc. <https://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 <https://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
-<https://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
-<https://www.gnu.org/licenses/why-not-lgpl.html>.
-

package.json

@@ -1,11 +1,10 @@
 {
   "name": "purescript-csv-stream",
-  "version": "v2.3.0",
+  "version": "v1.2.17",
   "type": "module",
   "dependencies": {
     "csv-parse": "^5.5.5",
-    "csv-stringify": "^6.4.6",
+    "csv-stringify": "^6.4.6"
-    "decimal.js": "^10.4.3"
   },
   "devDependencies": {
     "typescript": "^5.4.5"

spago.lock

@@ -4,11 +4,15 @@ workspace:
       path: ./
       dependencies:
         - aff: ">=7.1.0 <8.0.0"
+        - aff-promise
         - arrays: ">=7.3.0 <8.0.0"
         - bifunctors: ">=6.0.0 <7.0.0"
+        - control: ">=6.0.0 <7.0.0"
         - datetime: ">=6.1.0 <7.0.0"
         - effect: ">=4.0.0 <5.0.0"
+        - either: ">=6.1.0 <7.0.0"
         - exceptions: ">=6.0.0 <7.0.0"
+        - filterable: ">=5.0.0 <6.0.0"
         - foldable-traversable: ">=6.0.0 <7.0.0"
         - foreign: ">=7.0.0 <8.0.0"
         - foreign-object: ">=4.1.0 <5.0.0"
@@ -18,32 +22,27 @@ workspace:
         - newtype: ">=5.0.0 <6.0.0"
         - node-buffer: ">=9.0.0 <10.0.0"
         - node-event-emitter: ">=3.0.0 <4.0.0"
-        - node-stream-pipes: ">=2.1.0 <3.0.0"
         - node-streams: ">=9.0.0 <10.0.0"
         - nullable: ">=6.0.0 <7.0.0"
         - numbers: ">=9.0.1 <10.0.0"
         - ordered-collections: ">=3.2.0 <4.0.0"
         - precise-datetime: ">=7.0.0 <8.0.0"
         - prelude: ">=6.0.1 <7.0.0"
+        - profunctor
         - record: ">=4.0.0 <5.0.0"
         - record-extra: ">=5.0.1 <6.0.0"
         - st: ">=6.2.0 <7.0.0"
         - strings: ">=6.0.1 <7.0.0"
+        - tailrec: ">=6.1.0 <7.0.0"
         - transformers: ">=6.0.0 <7.0.0"
-        - tuples: ">=7.0.0 <8.0.0"
         - typelevel-prelude: ">=7.0.0 <8.0.0"
         - unsafe-coerce: ">=6.0.0 <7.0.0"
       test_dependencies:
         - console
-        - gen
-        - node-fs
-        - node-process
-        - node-zlib
-        - quickcheck
-        - simple-json
         - spec
       build_plan:
         - aff
+        - aff-promise
         - ansi
         - arraybuffer-types
         - arrays
@@ -62,6 +61,7 @@ workspace:
         - enums
         - exceptions
         - exists
+        - filterable
         - fixed-points
         - foldable-traversable
         - foreign
@@ -77,19 +77,13 @@ workspace:
         - invariant
         - js-date
         - lazy
-        - lcg
         - lists
         - maybe
         - mmorph
         - newtype
         - node-buffer
         - node-event-emitter
-        - node-fs
-        - node-path
-        - node-process
-        - node-stream-pipes
         - node-streams
-        - node-zlib
         - nonempty
         - now
         - nullable
@@ -100,17 +94,13 @@ workspace:
         - parsing
         - partial
         - pipes
-        - posix-types
         - precise-datetime
         - prelude
         - profunctor
-        - quickcheck
-        - random
         - record
         - record-extra
         - refs
         - safe-coerce
-        - simple-json
         - spec
         - st
         - strings
@@ -121,9 +111,7 @@ workspace:
         - typelevel-prelude
         - unfoldable
         - unicode
-        - unordered-collections
         - unsafe-coerce
-        - variant
   extra_packages: {}
 packages:
   aff:
@@ -148,6 +136,13 @@ packages:
       - tailrec
       - transformers
       - unsafe-coerce
+  aff-promise:
+    type: registry
+    version: 4.0.0
+    integrity: sha256-Kq5EupbUpXeUXx4JqGQE7/RTTz/H6idzWhsocwlEFhM=
+    dependencies:
+      - aff
+      - foreign
   ansi:
     type: registry
     version: 7.0.0
@@ -327,6 +322,17 @@ packages:
     integrity: sha256-A0JQHpTfo1dNOj9U5/Fd3xndlRSE0g2IQWOGor2yXn8=
     dependencies:
       - unsafe-coerce
+  filterable:
+    type: registry
+    version: 5.0.0
+    integrity: sha256-cCojJHRnTmpY1j1kegI4CFwghdQ2Fm/8dzM8IlC+lng=
+    dependencies:
+      - arrays
+      - either
+      - foldable-traversable
+      - identity
+      - lists
+      - ordered-collections
   fixed-points:
     type: registry
     version: 7.0.0
@@ -503,17 +509,6 @@ packages:
       - foldable-traversable
       - invariant
       - prelude
-  lcg:
-    type: registry
-    version: 4.0.0
-    integrity: sha256-h7ME5cthLfbgJOJdsZcSfFpwXsx4rf8YmhebU+3iSYg=
-    dependencies:
-      - effect
-      - integers
-      - maybe
-      - partial
-      - prelude
-      - random
   lists:
     type: registry
     version: 7.0.0
@@ -578,86 +573,6 @@ packages:
       - nullable
       - prelude
       - unsafe-coerce
-  node-fs:
-    type: registry
-    version: 9.1.0
-    integrity: sha256-TzhvGdrwcM0bazDvrWSqh+M/H8GKYf1Na6aGm2Qg4+c=
-    dependencies:
-      - datetime
-      - effect
-      - either
-      - enums
-      - exceptions
-      - functions
-      - integers
-      - js-date
-      - maybe
-      - node-buffer
-      - node-path
-      - node-streams
-      - nullable
-      - partial
-      - prelude
-      - strings
-      - unsafe-coerce
-  node-path:
-    type: registry
-    version: 5.0.0
-    integrity: sha256-pd82nQ+2l5UThzaxPdKttgDt7xlsgIDLpPG0yxDEdyE=
-    dependencies:
-      - effect
-  node-process:
-    type: registry
-    version: 11.2.0
-    integrity: sha256-+2MQDYChjGbVbapCyJtuWYwD41jk+BntF/kcOTKBMVs=
-    dependencies:
-      - effect
-      - foreign
-      - foreign-object
-      - maybe
-      - node-event-emitter
-      - node-streams
-      - posix-types
-      - prelude
-      - unsafe-coerce
-  node-stream-pipes:
-    type: registry
-    version: 2.1.0
-    integrity: sha256-pYBOQY4bGEZzI5UHsUxJAhsKqtmE6CC1sHmFqgj64V8=
-    dependencies:
-      - aff
-      - arrays
-      - console
-      - control
-      - datetime
-      - effect
-      - either
-      - exceptions
-      - foldable-traversable
-      - foreign-object
-      - fork
-      - lists
-      - maybe
-      - mmorph
-      - newtype
-      - node-buffer
-      - node-event-emitter
-      - node-fs
-      - node-path
-      - node-streams
-      - node-zlib
-      - now
-      - ordered-collections
-      - parallel
-      - pipes
-      - prelude
-      - st
-      - strings
-      - tailrec
-      - transformers
-      - tuples
-      - unordered-collections
-      - unsafe-coerce
   node-streams:
     type: registry
     version: 9.0.0
@@ -671,19 +586,6 @@ packages:
       - node-event-emitter
       - nullable
       - prelude
-  node-zlib:
-    type: registry
-    version: 0.4.0
-    integrity: sha256-kYSajFQFzWVg71l5/y4w4kXdTr5EJoqyV3D2RqmAjQ4=
-    dependencies:
-      - aff
-      - effect
-      - either
-      - functions
-      - node-buffer
-      - node-streams
-      - prelude
-      - unsafe-coerce
   nonempty:
     type: registry
     version: 7.0.0
@@ -803,13 +705,6 @@ packages:
       - tailrec
       - transformers
       - tuples
-  posix-types:
-    type: registry
-    version: 6.0.0
-    integrity: sha256-ZfFz8RR1lee/o/Prccyeut3Q+9tYd08mlR72sIh6GzA=
-    dependencies:
-      - maybe
-      - prelude
   precise-datetime:
     type: registry
     version: 7.0.0
@@ -850,45 +745,6 @@ packages:
       - newtype
       - prelude
       - tuples
-  quickcheck:
-    type: registry
-    version: 8.0.1
-    integrity: sha256-ZvpccKQCvgslTXZCNmpYW4bUsFzhZd/kQUr2WmxFTGY=
-    dependencies:
-      - arrays
-      - console
-      - control
-      - effect
-      - either
-      - enums
-      - exceptions
-      - foldable-traversable
-      - gen
-      - identity
-      - integers
-      - lazy
-      - lcg
-      - lists
-      - maybe
-      - newtype
-      - nonempty
-      - numbers
-      - partial
-      - prelude
-      - record
-      - st
-      - strings
-      - tailrec
-      - transformers
-      - tuples
-      - unfoldable
-  random:
-    type: registry
-    version: 6.0.0
-    integrity: sha256-CJ611a35MPCE7XQMp0rdC6MCn76znlhisiCRgboAG+Q=
-    dependencies:
-      - effect
-      - integers
   record:
     type: registry
     version: 4.0.0
@@ -922,20 +778,6 @@ packages:
     integrity: sha256-a1ibQkiUcbODbLE/WAq7Ttbbh9ex+x33VCQ7GngKudU=
     dependencies:
       - unsafe-coerce
-  simple-json:
-    type: registry
-    version: 9.0.0
-    integrity: sha256-K3RJaThqsszTd+TEklzZmAdDqvIHWgXIfKqlsoykU1c=
-    dependencies:
-      - arrays
-      - exceptions
-      - foreign
-      - foreign-object
-      - nullable
-      - prelude
-      - record
-      - typelevel-prelude
-      - variant
   spec:
     type: registry
     version: 7.6.0
@@ -1068,36 +910,8 @@ packages:
       - foldable-traversable
       - maybe
       - strings
-  unordered-collections:
-    type: registry
-    version: 3.1.0
-    integrity: sha256-H2eQR+ylI+cljz4XzWfEbdF7ee+pnw2IZCeq69AuJ+Q=
-    dependencies:
-      - arrays
-      - enums
-      - functions
-      - integers
-      - lists
-      - prelude
-      - record
-      - tuples
-      - typelevel-prelude
-      - unfoldable
   unsafe-coerce:
     type: registry
     version: 6.0.0
     integrity: sha256-IqIYW4Vkevn8sI+6aUwRGvd87tVL36BBeOr0cGAE7t0=
     dependencies: []
-  variant:
-    type: registry
-    version: 8.0.0
-    integrity: sha256-SR//zQDg2dnbB8ZHslcxieUkCeNlbMToapvmh9onTtw=
-    dependencies:
-      - enums
-      - lists
-      - maybe
-      - partial
-      - prelude
-      - record
-      - tuples
-      - unsafe-coerce

spago.yaml

@@ -1,7 +1,7 @@
 package:
   name: csv-stream
   publish:
-    version: '2.3.0'
+    version: '1.2.17'
     license: 'GPL-3.0-or-later'
     location:
       githubOwner: 'cakekindel'
@@ -10,12 +10,17 @@ package:
     strict: true
     pedanticPackages: true
   dependencies:
+    - aff-promise
+    - profunctor
     - aff: ">=7.1.0 <8.0.0"
     - arrays: ">=7.3.0 <8.0.0"
     - bifunctors: ">=6.0.0 <7.0.0"
+    - control: ">=6.0.0 <7.0.0"
     - datetime: ">=6.1.0 <7.0.0"
     - effect: ">=4.0.0 <5.0.0"
+    - either: ">=6.1.0 <7.0.0"
     - exceptions: ">=6.0.0 <7.0.0"
+    - filterable: ">=5.0.0 <6.0.0"
     - foldable-traversable: ">=6.0.0 <7.0.0"
     - foreign: ">=7.0.0 <8.0.0"
     - foreign-object: ">=4.1.0 <5.0.0"
@@ -25,7 +30,6 @@ package:
     - newtype: ">=5.0.0 <6.0.0"
     - node-buffer: ">=9.0.0 <10.0.0"
     - node-event-emitter: ">=3.0.0 <4.0.0"
-    - node-stream-pipes: ">=2.1.0 <3.0.0"
     - node-streams: ">=9.0.0 <10.0.0"
     - nullable: ">=6.0.0 <7.0.0"
     - numbers: ">=9.0.1 <10.0.0"
@@ -36,20 +40,14 @@ package:
     - record-extra: ">=5.0.1 <6.0.0"
     - st: ">=6.2.0 <7.0.0"
     - strings: ">=6.0.1 <7.0.0"
+    - tailrec: ">=6.1.0 <7.0.0"
     - transformers: ">=6.0.0 <7.0.0"
-    - tuples: ">=7.0.0 <8.0.0"
     - typelevel-prelude: ">=7.0.0 <8.0.0"
     - unsafe-coerce: ">=6.0.0 <7.0.0"
   test:
     main: Test.Main
     dependencies:
       - console
-      - gen
-      - node-fs
-      - node-process
-      - node-zlib
-      - quickcheck
-      - simple-json
       - spec
 workspace:
   extraPackages: {}

src/Data.CSV.Record.purs

@@ -40,7 +40,7 @@ instance (RowToList r (Cons k v tailrl), IsSymbol k, ReadCSV v, Lacks k tail, Co
   readCSVRecord cols vals = do
     let
       k = reflectSymbol (Proxy @k)
-    pos <- liftMaybe (pure $ ForeignError $ "reached end of row looking for column " <> k) $ Map.lookup k cols
+    pos <- liftMaybe (pure $ ForeignError $ "row too long; did not expect value " <> k) $ Map.lookup k cols
     let valraw = fromMaybe "" $ Array.index vals pos
     val <- readCSV @v valraw
     tail <- readCSVRecord @tail @tailrl cols vals

src/Node.Stream.CSV.Parse.js

@@ -1,7 +1,30 @@
-import { Parser } from "csv-parse";
+import { parse, Parser } from "csv-parse";
 
-/** @type {(s: import('csv-parse').Options) => () => Parser} */
+class ParserWithColumns extends Parser {
-export const makeImpl = (c) => () => new Parser(c);
+  /** @type {Array<string>} */
+  columns = [];
+  /** @type {Map<string, number> | null} */
+  columnsMap = null;
+}
 
-/** @type {(s: Parser) => () => Array<string> | null} */
+/** @type {(s: import('csv-parse').Options) => () => ParserWithColumns} */
+export const makeImpl = (c) => () => {
+  const parser = new ParserWithColumns(c);
+  parser.once("readable", () => {
+    parser.columns = parser.read();
+    parser.emit("columns", parser.columns);
+  });
+  return parser;
+};
+
+/** @type {(s: ParserWithColumns) => () => Array<string> | null} */
 export const readImpl = (p) => () => p.read();
+
+/** @type {(s: ParserWithColumns) => () => Array<string>} */
+export const columnsArrayImpl = (p) => () => p.columns;
+
+/** @type {(s: ParserWithColumns) => () => Map<string, number> | null} */
+export const columnsMapImpl = (p) => () => p.columnsMap;
+
+/** @type {(s: ParserWithColumns) => (m: Map<string, number>) => () => void} */
+export const setColumnsMapImpl = (p) => (m) => () => (p.columnsMap = m);

src/Node.Stream.CSV.Parse.purs

@@ -2,18 +2,41 @@ module Node.Stream.CSV.Parse where
 
 import Prelude hiding (join)
 
+import Control.Alt ((<|>))
+import Control.Monad.Error.Class (liftEither, liftMaybe)
+import Control.Monad.Except (runExcept)
+import Control.Monad.Except.Trans (catchError)
+import Control.Monad.Maybe.Trans (MaybeT(..), runMaybeT)
+import Control.Monad.ST.Class (liftST)
+import Control.Monad.Trans.Class (lift)
+import Data.Array as Array
+import Data.Array.ST as Array.ST
+import Data.Bifunctor (lmap)
+import Data.CSV.Record (class ReadCSVRecord, readCSVRecord)
+import Data.Either (Either(..))
+import Data.Filterable (filter)
+import Data.Map (Map)
+import Data.Map as Map
+import Data.Maybe (Maybe(..), isNothing)
 import Data.Nullable (Nullable)
+import Data.Nullable as Nullable
 import Effect (Effect)
+import Effect.Aff (Canceler(..), makeAff)
+import Effect.Aff.Class (class MonadAff, liftAff)
+import Effect.Class (liftEffect)
+import Effect.Exception (error)
 import Effect.Uncurried (mkEffectFn1)
 import Foreign (Foreign, unsafeToForeign)
 import Foreign.Object (Object)
-import Foreign.Object (union) as Object
+import Foreign.Object as Object
-import Node.Buffer (Buffer)
+import Node.Encoding (Encoding(..))
 import Node.EventEmitter (EventHandle(..))
+import Node.EventEmitter as Event
 import Node.EventEmitter.UtilTypes (EventHandle1)
 import Node.Stream (Read, Stream, Write)
-import Node.Stream.Object (Transform) as Object
+import Node.Stream as Stream
 import Prim.Row (class Union)
+import Prim.RowList (class RowToList)
 import Unsafe.Coerce (unsafeCoerce)
 
 data CSVRead
@@ -22,9 +45,12 @@ data CSVRead
 -- | into parsed purescript objects.
 -- |
 -- | The CSV contents may be piped into this stream
--- | as Buffer or String chunks.
+-- | as Buffer or String encoded chunks.
-type CSVParser :: Row Type -> Type
+-- |
-type CSVParser r = Stream (read :: Read, write :: Write, csv :: CSVRead | r)
+-- | Records can be read with `read` when `Node.Stream.readable`
+-- | is true.
+type CSVParser :: Row Type -> Row Type -> Type
+type CSVParser a r = Stream (read :: Read, write :: Write, csv :: CSVRead | r)
 
 -- | https://csv.js.org/parse/options/
 type Config r =
@@ -56,22 +82,128 @@ type Config r =
   )
 
 -- | Create a CSVParser
-make :: forall @config @missing @extra. Union config missing (Config extra) => { | config } -> Effect (CSVParser ())
+make :: forall @r rl @config @missing @extra. RowToList r rl => ReadCSVRecord r rl => Union config missing (Config extra) => { | config } -> Effect (CSVParser r ())
 make = makeImpl <<< unsafeToForeign <<< Object.union (recordToForeign { columns: false, cast: false, cast_date: false }) <<< recordToForeign
 
-toObjectStream :: CSVParser () -> Object.Transform Buffer (Array String)
+-- | Synchronously parse a CSV string
-toObjectStream = unsafeCoerce
+parse
+  :: forall @r rl @config missing extra m
+   . MonadAff m
+  => RowToList r rl
+  => ReadCSVRecord r rl
+  => Union config missing (Config extra)
+  => { | config }
+  -> String
+  -> m (Array { | r })
+parse config csv = do
+  stream <- liftEffect $ make @r @config @missing @extra config
+  void $ liftEffect $ Stream.writeString stream UTF8 csv
+  liftEffect $ Stream.end stream
+  readAll stream
+
+-- | Loop until the stream is closed, invoking the callback with each record as it is parsed.
+foreach
+  :: forall @r rl x m
+   . MonadAff m
+  => RowToList r rl
+  => ReadCSVRecord r rl
+  => CSVParser r x
+  -> ({ | r } -> Effect Unit)
+  -> m Unit
+foreach stream cb = do
+  alreadyHaveCols <- liftEffect $ getOrInitColumnsMap stream
+  when (isNothing alreadyHaveCols)
+    $ liftAff
+    $ makeAff \res -> do
+        stop <- flip (Event.once columnsH) stream $ const do
+          void $ getOrInitColumnsMap stream
+          res $ Right unit
+        pure $ Canceler $ const $ liftEffect stop
+
+  liftAff $ makeAff \res -> do
+    removeDataListener <- flip (Event.on dataH) stream \row ->
+      void
+        $ flip catchError (res <<< Left)
+        $ do
+            cols <- liftMaybe (error "unreachable") =<< getOrInitColumnsMap stream
+            record <- liftEither $ lmap (error <<< show) $ runExcept $ readCSVRecord @r @rl cols row
+            flip catchError (liftEffect <<< res <<< Left) (cb record)
+
+    removeEndListener <- flip (Event.once Stream.endH) stream (res $ Right unit)
+    removeErrorListener <- flip (Event.once Stream.errorH) stream (res <<< Left)
+
+    pure $ Canceler $ const $ liftEffect do
+      removeDataListener
+      removeEndListener
+      removeErrorListener
+
+-- | Reads a parsed record from the stream.
+-- |
+-- | Returns `Nothing` when either:
+-- | - The internal buffer of parsed records has been exhausted, but there will be more (`Node.Stream.readable` and `Node.Stream.closed` are both `false`)
+-- | - All records have been processed (`Node.Stream.closed` is `true`)
+read
+  :: forall @r rl a
+   . RowToList r rl
+  => ReadCSVRecord r rl
+  => CSVParser r a
+  -> Effect (Maybe { | r })
+read stream = runMaybeT do
+  cols <- MaybeT $ getOrInitColumnsMap stream
+  raw :: Array String <- MaybeT $ Nullable.toMaybe <$> readImpl stream
+  liftEither $ lmap (error <<< show) $ runExcept $ readCSVRecord @r @rl cols raw
+
+-- | Collect all parsed records into an array
+readAll
+  :: forall @r rl a m
+   . MonadAff m
+  => RowToList r rl
+  => ReadCSVRecord r rl
+  => CSVParser r a
+  -> m (Array { | r })
+readAll stream = do
+  records <- liftEffect $ liftST $ Array.ST.new
+  foreach stream $ void <<< liftEffect <<< liftST <<< flip Array.ST.push records
+  liftEffect $ liftST $ Array.ST.unsafeFreeze records
 
 -- | `data` event. Emitted when a CSV record has been parsed.
-dataH :: forall a. EventHandle1 (CSVParser a) (Array String)
+dataH :: forall r a. EventHandle1 (CSVParser r a) (Array String)
 dataH = EventHandle "data" mkEffectFn1
 
+-- | `columns` event. Emitted when the header row has been parsed.
+columnsH :: forall r a. EventHandle1 (CSVParser r a) (Array String)
+columnsH = EventHandle "columns" mkEffectFn1
+
 -- | FFI
 foreign import makeImpl :: forall r. Foreign -> Effect (Stream r)
 
 -- | FFI
 foreign import readImpl :: forall r. Stream r -> Effect (Nullable (Array String))
 
+-- | FFI
+foreign import columnsArrayImpl :: forall r. Stream r -> Effect (Array String)
+
+-- | FFI
+foreign import columnsMapImpl :: forall r. Stream r -> Effect (Nullable (Map String Int))
+
+-- | FFI
+foreign import setColumnsMapImpl :: forall r. Stream r -> Map String Int -> Effect Unit
+
+-- | FFI
+getOrInitColumnsMap :: forall r x. CSVParser r x -> Effect (Maybe (Map String Int))
+getOrInitColumnsMap s = runMaybeT do
+  cols :: Array String <- MaybeT $ filter (not <<< Array.null) <$> Just <$> columnsArrayImpl s
+  let
+    get = MaybeT $ Nullable.toMaybe <$> columnsMapImpl s
+    init = do
+      let
+        ixs = Array.range 0 (Array.length cols - 1)
+        assoc = Array.zip cols ixs
+        map = Map.fromFoldable assoc
+      lift $ setColumnsMapImpl s map
+      pure map
+  get <|> init
+
 -- | FFI
 recordToForeign :: forall r. Record r -> Object Foreign
 recordToForeign = unsafeCoerce

src/Node.Stream.CSV.Stringify.purs

@@ -2,16 +2,30 @@ module Node.Stream.CSV.Stringify where
 
 import Prelude
 
+import Control.Monad.Rec.Class (class MonadRec, whileJust)
+import Control.Monad.ST.Global as ST
+import Data.Array as Array
+import Data.Array.ST as Array.ST
 import Data.CSV.Record (class WriteCSVRecord, writeCSVRecord)
+import Data.Either (Either(..), blush)
+import Data.Foldable (class Foldable, fold)
+import Data.Maybe (Maybe(..))
 import Data.String.Regex (Regex)
+import Data.Traversable (for_)
 import Effect (Effect)
+import Effect.Aff (Canceler(..), makeAff)
+import Effect.Aff.Class (class MonadAff, liftAff)
+import Effect.Class (liftEffect)
 import Foreign (Foreign, unsafeToForeign)
 import Foreign.Object (Object)
-import Foreign.Object (union) as Object
+import Foreign.Object as Object
+import Node.EventEmitter as Event
 import Node.Stream (Read, Stream, Write)
-import Node.Stream.Object (Transform) as Object
+import Node.Stream as Stream
 import Prim.Row (class Union)
 import Prim.RowList (class RowToList)
+import Record.Extra (class Keys, keys)
+import Type.Prelude (Proxy(..))
 import Unsafe.Coerce (unsafeCoerce)
 
 data CSVWrite
@@ -24,8 +38,8 @@ data CSVWrite
 -- | Stringified rows are emitted on the `Readable` end as string
 -- | chunks, meaning it can be treated as a `Node.Stream.Readable`
 -- | that has had `setEncoding UTF8` invoked on it.
-type CSVStringifier :: Row Type -> Type
+type CSVStringifier :: Row Type -> Row Type -> Type
-type CSVStringifier r = Stream (read :: Read, write :: Write, csv :: CSVWrite | r)
+type CSVStringifier a r = Stream (read :: Read, write :: Write, csv :: CSVWrite | r)
 
 -- | https://csv.js.org/stringify/options/
 type Config r =
@@ -34,6 +48,7 @@ type Config r =
   , record_delimiter :: String
   , escape :: String
   , escape_formulas :: Boolean
+  , header :: Boolean
   , quote :: String
   , quoted :: Boolean
   , quoted_empty :: Boolean
@@ -48,36 +63,43 @@ foreign import writeImpl :: forall r. Stream r -> Array String -> Effect Unit
 recordToForeign :: forall r. Record r -> Object Foreign
 recordToForeign = unsafeCoerce
 
--- | Create a raw Transform stream that accepts chunks of `Array String`,
+-- | Create a CSVStringifier
--- | and transforms them into string CSV rows.
+make :: forall @r rl @config @missing @extra. Keys rl => RowToList r rl => WriteCSVRecord r rl => Union config missing (Config extra) => { | config } -> Effect (CSVStringifier r ())
--- |
+make = makeImpl <<< unsafeToForeign <<< Object.union (recordToForeign { columns: Array.fromFoldable $ keys (Proxy @r) }) <<< recordToForeign
--- | Requires an ordered array of column names.
-make
-  :: forall @config @missing @extra
-  . Union config missing (Config extra)
-  => Array String
-  -> { | config }
-  -> Effect (CSVStringifier ())
-make columns =
-  makeImpl
-  <<< unsafeToForeign
-  <<< Object.union (recordToForeign { columns, header: true })
-  <<< recordToForeign
 
--- | Convert the raw stream to a typed ObjectStream
+-- | Synchronously stringify a collection of records
-toObjectStream :: CSVStringifier () -> Object.Transform (Array String) String
+stringify :: forall @r rl f m @config missing extra. MonadAff m => MonadRec m => Keys rl => Foldable f => RowToList r rl => WriteCSVRecord r rl => Union config missing (Config extra) => { | config } -> f { | r } -> m String
-toObjectStream = unsafeCoerce
+stringify config records = do
+  stream <- liftEffect $ make @r @config @missing @extra config
+  liftEffect $ for_ records \r -> write stream r
+  liftEffect $ Stream.end stream
+  readAll stream
 
 -- | Write a record to a CSVStringifier.
 -- |
 -- | The record will be emitted on the `Readable` end
 -- | of the stream as a string chunk.
-write :: forall @r rl a. RowToList r rl => WriteCSVRecord r rl => CSVStringifier a -> { | r } -> Effect Unit
+write :: forall @r rl a. RowToList r rl => WriteCSVRecord r rl => CSVStringifier r a -> { | r } -> Effect Unit
 write s = writeImpl s <<< writeCSVRecord @r @rl
 
--- | Write a record to a CSVStringifier.
+-- | Loop until the stream is closed, invoking the callback with each chunk of stringified CSV text.
--- |
+foreach :: forall m r x. MonadAff m => MonadRec m => CSVStringifier r x -> (String -> m Unit) -> m Unit
--- | The record will be emitted on the `Readable` end
+foreach stream cb = whileJust do
--- | of the stream as a string chunk.
+  isReadable <- liftEffect $ Stream.readable stream
-writeRaw :: forall a. CSVStringifier a -> Array String -> Effect Unit
+  liftAff $ when (not isReadable) $ makeAff \res -> do
-writeRaw = writeImpl
+    stop <- flip (Event.once Stream.readableH) stream $ res $ Right unit
+    pure $ Canceler $ const $ liftEffect stop
+  whileJust do
+    s <- liftEffect $ (join <<< map blush) <$> Stream.readEither stream
+    for_ s cb
+    pure $ void s
+  isClosed <- liftEffect $ Stream.closed stream
+  pure $ if isClosed then Nothing else Just unit
+
+-- | Read the stringified chunks until end-of-stream, returning the entire CSV string.
+readAll :: forall r a m. MonadAff m => MonadRec m => CSVStringifier r a -> m String
+readAll stream = do
+  chunks <- liftEffect $ ST.toEffect $ Array.ST.new
+  foreach stream $ void <<< liftEffect <<< ST.toEffect <<< flip Array.ST.push chunks
+  chunks' <- liftEffect $ ST.toEffect $ Array.ST.unsafeFreeze chunks
+  pure $ fold chunks'

src/Node.Stream.Object.js

@@ -0,0 +1,535 @@
+import Stream from "stream";
+
+const DEBUG = process.env['NODEJS_OBJECT_STREAM_TRACE'] !== ''
+
+/** @type {(s: string) => void} */
+const log = m => DEBUG ? console.log(m) : undefined;
+
+let chainCount = 0
+let composeCount = 0
+let neverCount = 0
+let onceCount = 0
+let bindCount = 0
+let zipCount = 0
+let mapCount = 0
+let constCount = 0
+let fromPromiseCount = 0
+
+export class Never extends Stream.Readable {
+  constructor() {
+    super({objectMode: true})
+    this.id = neverCount++
+  }
+
+  _read() {
+    log(`Never {id: ${this.id}}#_read()`)
+    log(`  this.push(null)`)
+    this.push(null)
+  }
+}
+
+/** @template T */
+export class Once extends Stream.Duplex {
+  /** @param {T} a */
+  constructor(a) {
+    super({objectMode: true, allowHalfOpen: false})
+    this.a = a
+    this.id = onceCount++
+    this.push(this.a)
+    this.push(null)
+    log(`Once {id: ${this.id}}#new()`)
+    log(`  this.push(${a})`)
+    log(`  this.push(null)`)
+  }
+
+  /** @type {Stream.Duplex['_write']} */
+  _write(_ck, _enc, cb) {
+    cb()
+  }
+
+  _read() {
+    log(`Once {id: ${this.id}}#_read()`)
+  }
+}
+
+/** @template T */
+export class Const extends Stream.Transform {
+  /** @param {T} a */
+  constructor(a) {
+    super({objectMode: true})
+    this.a = a
+    this.id = constCount++
+  }
+
+  /** @type {Stream.Transform['_transform']} */
+  _transform(_c, _enc, cb) {
+    log(`Const {id: ${this.id}}#_transform(${_c}, _, _)`)
+    log(`  cb(${this.a})`)
+    this.push(this.a)
+    cb()
+  }
+}
+
+/** @template T */
+export class FromPromise extends Stream.Readable {
+  /** @param {Promise<T>} p */
+  constructor(p) {
+    super({objectMode: true})
+    this.id = fromPromiseCount++
+    p
+       .then(a => {
+         log(`FromPromise {id: ${this.id}}#new()`)
+         log(`  ...p.then(...)`)
+         log(`    this.push(${a})`)
+         log(`    this.push(null)`)
+         this.push(a)
+         this.push(null)
+       })
+       .catch(e => {
+         log(`FromPromise {id: ${this.id}}#new()`)
+         log(`  ...p.catch(...)`)
+         log(`    this.destroy(${e})`)
+         this.destroy(e)
+       })
+  }
+
+  _read() {
+    log(`FromPromise {id: ${this.id}}#_read()`)
+  }
+}
+
+/** @template T */
+export class Chain extends Stream.Readable {
+  /** @param {...Stream.Readable} streams */
+  constructor(...streams) {
+    super({objectMode: true})
+    this.id = chainCount++
+    this.ix = -1
+    this.streams = streams
+    this.next()
+  }
+
+  next() {
+    log(`Chain {id: ${this.id}}#next()`)
+    this.ix++
+    if (this.ix === this.streams.length) {
+      log(`  this.push(null)`)
+      this.push(null)
+    } else {
+      const cur = this.streams[this.ix]
+
+      cur.once('error', e => {
+        log(`Chain {id: ${this.id}}#next()`)
+        log(`  cur.once('error', ...)`)
+        log(`    this.destroy(${e})`)
+        this.destroy(e)
+      })
+
+      cur.once('end', () => {
+        log(`Chain {id: ${this.id}}#next()`)
+        log(`  cur.once('end', ...)`)
+        log(`    this.next()`)
+        this.next()
+      })
+
+      cur.on('data', ck => {
+        log(`Chain {id: ${this.id}}#next()`)
+        log(`  cur.on('data', ...)`)
+        log(`    this.push(${ck})`)
+        const canPush = this.push(ck)
+        if (cur && !canPush) {
+          log(`    cur.pause()`)
+          cur.pause()
+        }
+      })
+    }
+  }
+
+  _read() {
+    log(`Chain {id: ${this.id}}#_read()`)
+    this.streams.forEach(s => {
+      if (s.isPaused()) {
+        log(`  s.resume()`)
+        s.resume()
+      }
+    })
+  }
+}
+
+/**
+  * @template T
+  * @template R
+  */
+export class Map extends Stream.Transform {
+  /** @param {(t: T) => R} f */
+  constructor(f) {
+    super({objectMode: true})
+    this.f = f
+    this.id = mapCount++
+  }
+
+  /** @type {Stream.Transform['_transform']} */
+  _transform(ck, _, cb) {
+    log(`Map {id: ${this.id}}#_transform(${ck}, _, _)`)
+    const r = this.f(ck)
+    log(`  const r = (${r})`)
+    log(`  cb(null, ${r})`)
+    cb(null, r)
+  }
+}
+
+export class Zip extends Stream.Readable {
+  /** @type {Array<Stream.Readable>} */
+  streams = []
+
+  /** @type {Array<Array<unknown | null>>} */
+  buf = []
+
+  /** @type {Timer | null} */
+  readingTimer = null
+  /** @type {Timer | null} */
+  pushingTimer = null
+
+  /** @param {...Stream.Readable} streams */
+  constructor(...streams) {
+    super({objectMode: true})
+    this.id = zipCount++
+    this.streams = streams
+    this.streams.forEach((s, ix) => {
+      s.once('error', e => this.destroy(e))
+      s.once('end', () => this.push(null))
+      s.on('data', ck => {
+        this.bufput(ix, ck)
+        s.pause()
+      })
+    })
+  }
+
+  /** @type {(ix: number, val: unknown) => void} */
+  bufput(ix, val) {
+    if (this.buflastdone()) {
+      this.bufnew()
+    }
+    const zipped = this.buflast()
+    zipped[ix] = val
+  }
+
+  bufnew() {
+    this.buf.push(this.streams.map(_ => null))
+  }
+
+  buflastdone() {
+    const l = this.buflast()
+    return l.every(a => a !== null)
+  }
+
+  bufheaddone() {
+    const l = this.bufhead()
+    return l.every(a => a !== null)
+  }
+
+  /** @returns {unknown[]} */
+  buflast() {
+    if (this.buf.length === 0) {
+      this.bufnew()
+    }
+    return this.buf[this.buf.length - 1]
+  }
+
+  /** @returns {unknown[]} */
+  bufhead() {
+    if (this.buf.length === 0) {
+      this.bufnew()
+    }
+    return this.buf[0]
+  }
+
+  anyended() {
+    return this.streams.some(s => s.readableEnded)
+  }
+
+  maybeend() {
+    if (!this.bufheaddone() && this.anyended()) {
+      this.push(null)
+      return true
+    } else {
+      return false
+    }
+  }
+
+  readzipped() {
+    if (this.anyended()) {
+      return false
+    }
+
+    if (this.buflastdone()) {
+      this.bufnew()
+      this.streams.forEach((s, ix) => {
+        this.buflast()[ix] = s.read()
+      })
+      return this.buflastdone()
+    } else {
+      const missingIxs = this.buflast().flatMap((a, ix) => a === null ? [ix] : [])
+      missingIxs.forEach(ix => {
+        this.buflast()[ix] = this.streams[ix].read()
+      })
+      return this.buflastdone()
+    }
+  }
+
+  _read() {
+    if (this.readingTimer) clearTimeout(this.readingTimer);
+    if (this.pushingTimer) clearTimeout(this.pushingTimer);
+
+    this.readingTimer = setTimeout(() => {
+      while (this.readzipped()) {
+        continue;
+      }
+    }, 0)
+
+    this.pushingTimer = setTimeout(() => {
+      let canWrite = true
+
+      if (this.maybeend()) {
+        return
+      }
+
+      while (canWrite && this.bufheaddone()) {
+        canWrite = this.push(this.bufhead())
+        this.buf.shift()
+      }
+
+      this.maybeend()
+    }, 0)
+  }
+}
+
+export class Compose extends Stream.Duplex {
+  /**
+   * @param {Stream.Readable | Stream.Transform} a
+   * @param {Stream.Transform} b
+   */
+  constructor(a, b) {
+    super({objectMode: true})
+    this.id = composeCount++
+
+    this.a = a
+    this.b = b
+
+    log(`Compose {id: ${this.id}}#new()`)
+    log(`  a.on('data', ...)`)
+    log(`  a.once('end', ...)`)
+    log(`  a.once('error', ...)`)
+    log(`  a.pause()`)
+    log(`  b.on('drain', ...)`)
+    log(`  b.on('data', ...)`)
+    log(`  b.on('error', ...)`)
+    log(`  b.on('finish', ...)`)
+
+    this.a.once('end', () => {
+      log(`Compose {id: ${this.id}}#new()`)
+      log(`  a.on('end', ...)`)
+      log(`    b.end()`)
+      this.b.end()
+    })
+
+    this.a.on('data', ck => {
+      log(`Compose {id: ${this.id}}#new()`)
+      log(`  a.on('data', ...)`)
+      log(`    b.write(${ck})`)
+      const canWrite = this.b.write(ck)
+
+      if (!canWrite) {
+        log(`    a.pause()`)
+        this.a.pause()
+      }
+    })
+
+    this.a.once('error', e => {
+      log(`Compose {id: ${this.id}}#new()`)
+      log(`  a.once('error', ...)`)
+      log(`    this.destroy(${e})`)
+      this.destroy(e)
+      this.b.destroy(e)
+    })
+
+    this.b.on('drain', () => {
+      log(`Compose {id: ${this.id}}#new()`)
+      log(`  b.on('drain', ...)`)
+      log(`    this.a.resume()`)
+      this.a.resume()
+    })
+
+    this.b.on('data', ck => {
+      log(`Compose {id: ${this.id}}#new()`)
+      log(`  b.on('data', ...)`)
+      log(`    this.push(${ck})`)
+      const canPush = this.push(ck)
+      if (!canPush) {
+        log(`    b.pause()`)
+        this.b.pause()
+      }
+    })
+
+    this.b.once('error', e => {
+      log(`Compose {id: ${this.id}}#new()`)
+      log(`  b.once('error', ...)`)
+      log(`    this.destroy(${e})`)
+      this.destroy(e)
+      this.a.destroy(e)
+    })
+
+    this.b.once('finish', () => {
+      log(`Compose {id: ${this.id}}#new()`)
+      log(`  b.once('finish', ...)`)
+      log(`    this.emit('finish')`)
+      this.push(null)
+      this.end()
+      this.emit('finish')
+    })
+  }
+
+  _read() {
+    log(`Compose {id: ${this.id}}#_read()`)
+    if (this.b.isPaused()) {
+      log(`  b.resume()`)
+      this.b.resume()
+    }
+  }
+
+  /** @type {Stream.Duplex['_write']} */
+  _write(ck, _enc, cb) {
+    log(`Compose {id: ${this.id}}#_write(${ck}, _, _)`)
+    if (this.a instanceof Stream.Readable) {
+      throw new Error('Cannot `write` to a Readable stream')
+    }
+
+    log(`  this.a.write(${ck}, _, _)`)
+    this.a.write(ck, _enc, cb)
+  }
+}
+
+/**
+ * @template T 
+ */
+export class Bind extends Stream.Duplex {
+  /**
+   * @param {(t: T) => () => Stream.Readable} f
+   */
+  constructor(f) {
+    super({objectMode: true, allowHalfOpen: true})
+    this.f = f
+    this.id = bindCount++
+
+    this.ix = 0
+
+    /** @type {Array<Stream.Readable>} */
+    this.streams = []
+
+    /** @type {(() => void) | undefined} */
+    this.doneCb = undefined
+
+    this.paused = true
+  }
+
+  /** @type {NonNullable<Stream.Duplex['_final']>} */
+  _final(cb) {
+    log(`Bind {id: ${this.id}}#_final(_)`)
+    this.doneCb = cb
+  }
+
+  init() {
+    log(`Bind {id: ${this.id}}#init()`)
+
+    const s = this.streams[this.ix]
+    if (this.paused || (!s && !this.doneCb)) {
+      this.paused = true
+      return
+    } else if (this.doneCb) {
+      log(`  this.doneCb()`)
+      this.doneCb()
+      this.doneCb = undefined
+      return
+    }
+
+    log(`  s.on('data', ...)`)
+    s.on('data', ck => {
+      log(`Bind {id: ${this.id}}#initcur()`)
+      log(`  s.on('data', ...)`)
+      log(`    this.push(${ck})`)
+      const canPush = this.push(ck)
+      if (!canPush) {
+        s.pause()
+      }
+    })
+
+    log(`  s.once('end', ...)`)
+    s.once('end', () => {
+      log(`Bind {id: ${this.id}}#initcur()`)
+      log(`  s.once('end', ...)`)
+      log(`    this.ix++`)
+      log(`    this.init()`)
+      this.ix++
+      this.init()
+    })
+  }
+
+  /** @type {Stream.Duplex['_write']} */
+  _write(ck, _, cb) {
+    log(`Bind {id: ${this.id}}#_write(${ck}, _, _)`)
+    try {
+      log(`  this.streams = ${JSON.stringify(this.streams.map(_ => 'Readable'))}`)
+      this.streams.push(this.f(ck)())
+      if (this.paused) {
+        log(`  this.init()`)
+        this.paused = false
+        this.init()
+      }
+      log(`  cb()`)
+      cb()
+    } catch(e) {
+      log(`  cb(${e})`)
+      // @ts-ignore
+      cb(e)
+    }
+  }
+
+  /** @type {Stream.Duplex['_read']} */
+  _read() {
+    log(`Bind {id: ${this.id}}#_read()`)
+    this.streams.forEach(s => 
+    s.isPaused() ? s.resume() : undefined)
+  }
+}
+
+/** @type {() => Stream.Readable} */
+export const neverImpl = () => new Never();
+
+/** @type {<T>(a: T) => () => Stream.Transform} */
+export const constImpl = (a) => () => new Const(a);
+
+/** @type {<T>(a: T) => () => Stream.Readable} */
+export const onceImpl = (a) => () => new Once(a);
+
+/** @type {<T>(a: () => Promise<T>) => () => Stream.Readable} */
+export const fromPromiseImpl = (a) => () => new FromPromise(a())
+
+/** @type {(ss: Array<Stream.Readable>) => () => Stream.Readable} */
+export const chainImpl = (ss) => () => new Chain(...ss);
+
+/** @type {<T, R>(f: (t: T) => R) => () => Stream.Transform} */
+export const mapImpl = (f) => () => new Map(f);
+
+/** @type {(iab: Stream.Readable) => (ia: Stream.Readable) => () => Stream.Readable} */
+export const applyImpl = (iab) => (ia) => () =>
+  new Compose(new Zip(iab, ia), new Map(([ab, a]) => ab(a)))
+
+/** @type {<T>(f: (t: T) => () => Stream.Readable) => () => Stream.Duplex} */
+export const bindImpl = (f) => () => new Bind(f)
+
+/** @type {(a: Stream.Transform | Stream.Readable) => (b: Stream.Transform) => () => Stream.Duplex} */
+export const pipeImpl = (a) => (b) => () => new Compose(a, b)
+
+process.on('beforeExit', () => {
+  debugger;
+})

src/Node.Stream.Object.purs

@@ -0,0 +1,189 @@
+module Node.Stream.Object where
+
+import Prelude
+
+import Control.Monad.Rec.Class (untilJust)
+import Control.Monad.ST.Global as ST
+import Control.Monad.ST.Ref as STRef
+import Control.Promise (Promise)
+import Control.Promise as Promise
+import Data.Array as Array
+import Data.Array.ST as Array.ST
+import Data.Either (Either(..))
+import Data.Maybe (Maybe(..))
+import Data.Newtype (class Newtype, unwrap, wrap)
+import Data.Profunctor (class Profunctor)
+import Data.Traversable (class Traversable, traverse)
+import Effect (Effect)
+import Effect.Aff (Aff, delay, effectCanceler, launchAff_, makeAff)
+import Effect.Aff.Class (class MonadAff)
+import Effect.Class (class MonadEffect, liftEffect)
+import Effect.Exception (Error)
+import Effect.Uncurried (mkEffectFn1)
+import Node.Buffer (Buffer)
+import Node.EventEmitter (EventHandle(..))
+import Node.EventEmitter as Event
+import Node.EventEmitter.UtilTypes (EventHandle1, EventHandle0)
+import Node.Stream (Readable, Writable)
+import Unsafe.Coerce (unsafeCoerce)
+
+foreign import data Stream :: Type -> Type -> Type
+
+newtype ObjectStream :: Type -> Type -> Type
+newtype ObjectStream i o = ObjectStream (Effect (Stream i o))
+
+derive instance Newtype (ObjectStream i o) _
+
+instance Functor (ObjectStream i) where
+  map :: forall a b. (a -> b) -> ObjectStream i a -> ObjectStream i b
+  map ab (ObjectStream ia) = wrap $ join $ pure pipeImpl <*> ia <*> mapImpl ab
+
+instance Apply (ObjectStream i) where
+  apply :: forall a b. ObjectStream i (a -> b) -> ObjectStream i a -> ObjectStream i b
+  apply (ObjectStream iab) (ObjectStream ia) = wrap $ join $ pure applyImpl <*> iab <*> ia
+
+instance Applicative (ObjectStream i) where
+  pure = once
+
+instance Monad (ObjectStream i)
+
+instance MonadEffect (ObjectStream Unit) where
+  liftEffect = wrap <<< flip bind onceImpl
+
+instance MonadAff (ObjectStream Unit) where
+  liftAff = wrap <<< fromPromiseImpl <<< Promise.fromAff
+
+instance Bind (ObjectStream i) where
+  bind (ObjectStream sia') asb = wrap do
+    sia <- sia'
+    sab <- bindImpl (unwrap <<< asb)
+    pipeImpl sia sab
+
+instance Profunctor ObjectStream where
+  dimap ab cd (ObjectStream sbc') = wrap do
+    sbc <- sbc'
+    sab <- mapImpl ab
+    scd <- mapImpl cd
+    sac <- pipeImpl sab sbc
+    pipeImpl sac scd
+
+-- | A stream that will emit the value `a` exactly once, and ignores any written chunks.
+once :: forall i a. a -> ObjectStream i a
+once = wrap <<< onceImpl
+
+-- | A stream that will immediately emit `close` and `end` events.
+never :: forall a. ObjectStream Unit a
+never = wrap neverImpl
+
+-- | A stream that for all input chunks, emits `unit`.
+sink :: forall a. ObjectStream a Unit
+sink = pure unit
+
+-- | Create a stream from a `Foldable` of `a`s
+fromFoldable :: forall f a. Traversable f => f a -> ObjectStream Unit a
+fromFoldable = chainMany <<< map once
+
+-- | Convert a `Readable` stream emitting `Buffer` chunks to an `ObjectStream`
+fromBufferReadable :: forall r. Readable r -> ObjectStream Unit Buffer
+fromBufferReadable r = wrap $ pure $ (unsafeCoerce :: Readable r -> Stream Unit Buffer) r
+
+-- | Convert a `Readable` stream emitting `String` chunks to an `ObjectStream`
+fromStringReadable :: forall r. Readable r -> ObjectStream Unit String
+fromStringReadable r = wrap $ pure $ (unsafeCoerce :: Readable r -> Stream Unit String) r
+
+-- | Convert a `Writable` stream accepting `Buffer` chunks to an `ObjectStream`
+fromBufferWritable :: forall r. Writable r -> ObjectStream Buffer Unit
+fromBufferWritable r = wrap $ pure $ (unsafeCoerce :: Writable r -> Stream Buffer Unit) r
+
+-- | Convert a `Writable` stream accepting `String` chunks to an `ObjectStream`
+fromStringWritable :: forall r. Writable r -> ObjectStream String Unit
+fromStringWritable r = wrap $ pure $ (unsafeCoerce :: Writable r -> Stream String Unit) r
+
+-- | Emit chunks from the first stream, then when exhausted emit chunks from the second.
+chain :: forall a. ObjectStream Unit a -> ObjectStream Unit a -> ObjectStream Unit a
+chain a b = chainMany [ a, b ]
+
+-- | `chain` for an arbitrary number of streams.
+chainMany :: forall f a. Traversable f => f (ObjectStream Unit a) -> ObjectStream Unit a
+chainMany as' = wrap do
+  as <- Array.fromFoldable <$> traverse unwrap as'
+  chainImpl as
+
+run_ :: forall a. ObjectStream Unit a -> Aff Unit
+run_ = void <<< run
+
+run :: forall a. ObjectStream Unit a -> Aff (Array a)
+run (ObjectStream s') = do
+  runningCount <- liftEffect $ ST.toEffect $ STRef.new 0
+  values <- liftEffect $ ST.toEffect $ Array.ST.new
+  s <- liftEffect s'
+  makeAff \res ->
+    let
+      onData a = ST.toEffect do
+        void $ STRef.modify (_ + 1) runningCount
+        void $ Array.ST.push a values
+        void $ STRef.modify (_ - 1) runningCount
+      onError e = res $ Left e
+      onEnd = launchAff_ do
+        untilJust do
+          delay $ wrap $ 1.0
+          running <- liftEffect $ ST.toEffect $ STRef.read runningCount
+          pure $ if running == 0 then Just unit else Nothing
+        values' <- liftEffect $ ST.toEffect $ Array.ST.unsafeFreeze values
+        liftEffect $ res $ Right values'
+    in
+      do
+        cancelData <- Event.on dataH onData s
+        cancelError <- Event.on errorH onError s
+        cancelEnd <- Event.on endH onEnd s
+        pure $ effectCanceler do
+          cancelData
+          cancelError
+          cancelEnd
+
+-- | Constructs a Stream that re-emits the outputs from each stream, in order.
+foreign import chainImpl :: forall a. Array (Stream Unit a) -> Effect (Stream Unit a)
+
+-- | Pipes a stream's output into another's input, returning the new composite stream.
+-- |
+-- | Note that this differs from `Readable#pipe`, which returns the destination stream
+-- | verbatim to allow chained piping of only _outputs_.
+foreign import pipeImpl :: forall a b c. Stream a b -> Stream b c -> Effect (Stream a c)
+
+-- | A readable stream that immediately closes without emitting any chunks.
+foreign import neverImpl :: forall a. Effect (Stream Unit a)
+
+-- | Constructs a readable stream from an asynchronous value.
+foreign import fromPromiseImpl :: forall a. Effect (Promise a) -> Effect (Stream Unit a)
+
+-- | Constructs a readable stream that emits a single value then closes.
+foreign import onceImpl :: forall i a. a -> Effect (Stream i a)
+
+-- | Constructs a `Transform` applying the given function to chunks.
+foreign import mapImpl :: forall a b. (a -> b) -> Effect (Stream a b)
+
+-- | Constructs a `Transform` zipping functions from the left stream with values from the right stream.
+-- |
+-- | Closes when either stream closes.
+foreign import applyImpl :: forall i a b. Stream i (a -> b) -> Stream i a -> Effect (Stream i b)
+
+-- | Constructs a `Transform` which applies a function to each written chunk.
+-- |
+-- | The values emitted by the stream returned by this function are then emitted
+-- | until the temporary stream closes.
+foreign import bindImpl :: forall a _x b. (a -> Effect (Stream _x b)) -> Effect (Stream a b)
+
+dataH :: forall i o. EventHandle1 (Stream i o) o
+dataH = EventHandle "data" mkEffectFn1
+
+readableH :: forall i o. EventHandle0 (Stream i o)
+readableH = EventHandle "readable" identity
+
+closeH :: forall i o. EventHandle0 (Stream i o)
+closeH = EventHandle "close" identity
+
+endH :: forall i o. EventHandle0 (Stream i o)
+endH = EventHandle "end" identity
+
+errorH :: forall i o. EventHandle1 (Stream i o) Error
+errorH = EventHandle "error" mkEffectFn1

src/Pipes.CSV.purs

@@ -1,82 +0,0 @@
-module Pipes.CSV where
-
-import Prelude
-
-import Control.Monad.Cont (lift)
-import Control.Monad.Error.Class (liftEither, liftMaybe)
-import Control.Monad.Except (runExcept)
-import Control.Monad.ST.Global as ST
-import Control.Monad.ST.Ref as STRef
-import Data.Array as Array
-import Data.Bifunctor (lmap)
-import Data.CSV.Record (class ReadCSVRecord, class WriteCSVRecord, readCSVRecord, writeCSVRecord)
-import Data.FunctorWithIndex (mapWithIndex)
-import Data.Map as Map
-import Data.Maybe (Maybe(..))
-import Data.Tuple.Nested ((/\))
-import Effect.Aff (Aff)
-import Effect.Class (liftEffect)
-import Effect.Exception (error)
-import Node.Buffer (Buffer)
-import Node.Stream.CSV.Parse as CSV.Parse
-import Node.Stream.CSV.Stringify as CSV.Stringify
-import Pipes.Async (AsyncPipe(..), ReadResult(..), getAsyncIO, mapIO, stripIO)
-import Pipes.Node.Stream as Pipes.Stream
-import Prim.RowList (class RowToList)
-import Record.Extra (class Keys, keys)
-import Type.Prelude (Proxy(..))
-
--- | Transforms buffer chunks of a CSV file to parsed
--- | records of `r`.
-parse
-  :: forall @r rl
-  . RowToList r rl
-  => ReadCSVRecord r rl
-  => AsyncPipe (Maybe Buffer) (Maybe { | r }) Aff Unit
-parse =
-  let
-    readCols st = liftEffect $ ST.toEffect $ STRef.read st
-    putCols st a = void $ liftEffect $ ST.toEffect $ STRef.write (Just a) st
-
-    firstRow st a = putCols st $ Map.fromFoldable $ mapWithIndex (flip (/\)) a
-    row a cols' = liftEither $ lmap (error <<< show) $ runExcept $ readCSVRecord @r @rl cols' a
-  in do
-    colsST <- liftEffect $ ST.toEffect $ STRef.new Nothing
-
-    let
-      decoder = Pipes.Stream.fromTransformEffect $ CSV.Parse.toObjectStream <$> CSV.Parse.make {}
-
-    {write, read: read', awaitRead, awaitWrite} /\ done <- liftMaybe (error "unreachable") =<< lift (getAsyncIO decoder)
-    let
-      read =
-        read' >>= case _ of
-          ReadWouldBlock -> pure ReadWouldBlock
-          ReadOk Nothing -> pure $ ReadOk Nothing
-          ReadOk (Just r') -> do
-            cols <- readCols colsST
-            case cols of
-              Just cols' -> ReadOk <$> Just <$> row r' cols'
-              Nothing -> firstRow colsST r' *> read
-
-    AsyncIO $ {write, read, awaitWrite, awaitRead} /\ lift (stripIO done)
-
--- | Transforms buffer chunks of a CSV file to parsed
--- | arrays of CSV values.
-parseRaw :: AsyncPipe (Maybe Buffer) (Maybe (Array String)) Aff Unit
-parseRaw =
-  Pipes.Stream.fromTransformEffect $ CSV.Parse.toObjectStream <$> CSV.Parse.make {}
-
--- | Transforms CSV rows into stringified CSV records
--- | using the given ordered array of column names.
-stringifyRaw :: Array String -> AsyncPipe (Maybe (Array String)) (Maybe String) Aff Unit
-stringifyRaw columns =
-  Pipes.Stream.fromTransformEffect $ CSV.Stringify.toObjectStream <$> CSV.Stringify.make columns {}
-
--- | Transforms purescript records into stringified CSV records.
--- |
--- | Columns are inferred from the record's keys, ordered alphabetically.
-stringify :: forall r rl. WriteCSVRecord r rl => RowToList r rl => Keys rl => AsyncPipe (Maybe { | r }) (Maybe String) Aff Unit
-stringify = do
-  let
-    p = Pipes.Stream.fromTransformEffect $ CSV.Stringify.toObjectStream <$> CSV.Stringify.make (Array.fromFoldable $ keys $ Proxy @r) {}
-  mapIO (map $ writeCSVRecord @r @rl) identity p

test/Test/Main.purs

@@ -5,16 +5,10 @@ import Prelude
 import Data.Maybe (Maybe(..))
 import Effect (Effect)
 import Effect.Aff (launchAff_)
-import Effect.Class (liftEffect)
+import Test.Node.Stream.Object as Test.Node.Stream.Object
-import Effect.Console as Console
+import Test.Spec.Reporter (consoleReporter, specReporter)
-import Node.EventEmitter as Event
-import Node.Process as Process
-import Test.Pipes.CSV as Test.Pipes.CSV
-import Test.Spec.Reporter (specReporter)
 import Test.Spec.Runner (defaultConfig, runSpec')
 
 main :: Effect Unit
-main = launchAff_ do
+main = launchAff_ $ runSpec' (defaultConfig { timeout = Nothing }) [ consoleReporter ] do
-  void $ liftEffect $ Event.on Process.uncaughtExceptionH (const $ Console.error) Process.process
+  Test.Node.Stream.Object.spec
-  runSpec' (defaultConfig { failFast = true, timeout = Nothing }) [ specReporter ] do
-    Test.Pipes.CSV.spec

test/Test/Node.Stream.Object.purs

@@ -0,0 +1,57 @@
+module Test.Node.Stream.Object where
+
+import Prelude
+
+import Data.Newtype (wrap)
+import Effect.Aff (delay)
+import Effect.Aff.Class (liftAff)
+import Node.Stream.Object as Stream
+import Test.Spec (Spec, describe, it)
+import Test.Spec.Assertions (shouldEqual)
+
+spec :: Spec Unit
+spec =
+  describe "Node.Stream.Object" do
+    describe "ObjectStream" do
+      describe "once" do
+        it "emits once" do
+          out <- Stream.run (Stream.once 1)
+          out `shouldEqual` [ 1 ]
+      describe "never" do
+        it "immediately closes" do
+          out <- Stream.run Stream.never
+          out `shouldEqual` ([] :: Array Int)
+      describe "chain" do
+        it "noops" do
+          out <- Stream.run $ Stream.chainMany []
+          out `shouldEqual` ([] :: Array Int)
+        it "works with 1 stream" do
+          out <- Stream.run $ Stream.chainMany [Stream.once 1]
+          out `shouldEqual` [1]
+        it "works with 2 streams" do
+          out <- Stream.run $ Stream.chainMany [Stream.once 1, Stream.once 2]
+          out `shouldEqual` [1, 2]
+        it "does not emit end until last child stream ends" do
+          let
+            delayed n a = liftAff do
+              delay $ wrap n
+              pure a
+          out <- Stream.run $ Stream.chainMany [delayed 10.0 1, delayed 20.0 2]
+          out `shouldEqual` [1, 2]
+      describe "fromFoldable" do
+        it "creates an empty readable" do
+          out <- Stream.run (Stream.fromFoldable [] :: Stream.ObjectStream Unit (Array Int))
+          out `shouldEqual` []
+        it "creates a readable that emits each element" do
+          out <- Stream.run (Stream.fromFoldable [ 1, 2, 3 ])
+          out `shouldEqual` [ 1, 2, 3 ]
+        it "bind maps each number" do
+          out <- Stream.run do
+            a <- Stream.fromFoldable [ 1, 2, 3 ]
+            pure $ a + 1
+          out `shouldEqual` [ 2, 3, 4 ]
+        it "bind fans out" do
+          out <- Stream.run do
+            a <- Stream.fromFoldable [ 1, 2, 3 ]
+            Stream.fromFoldable [a * 10, a * 20]
+          out `shouldEqual` [ 10, 20, 20, 40, 30, 60 ]

test/Test/Pipes.CSV.purs

@@ -1,89 +0,0 @@
-module Test.Pipes.CSV where
-
-import Prelude
-
-import Control.Monad.Gen (chooseInt)
-import Control.Monad.Rec.Class (Step(..), tailRecM)
-import Data.Array as Array
-import Data.DateTime (DateTime)
-import Data.Foldable (fold, sum)
-import Data.Maybe (Maybe(..), fromJust)
-import Data.Newtype (wrap)
-import Data.PreciseDateTime (fromRFC3339String, toDateTimeLossy)
-import Data.String.CodePoints as String.CodePoints
-import Data.Tuple.Nested ((/\))
-import Effect.Class (liftEffect)
-import Effect.Console (log)
-import Node.Encoding (Encoding(..))
-import Partial.Unsafe (unsafePartial)
-import Pipes (yield, (>->))
-import Pipes.Async ((>-/->))
-import Pipes.CSV as Pipes.CSV
-import Pipes.Collect as Pipes.Collect
-import Pipes.Construct as Pipes.Construct
-import Pipes.Node.Buffer as Pipes.Buffer
-import Pipes.Node.Stream as Pipes.Stream
-import Pipes.Prelude (chain, map, toListM) as Pipes
-import Pipes.Util as Pipes.Util
-import Test.QuickCheck.Gen (randomSample')
-import Test.Spec (Spec, before, describe, it)
-import Test.Spec.Assertions (shouldEqual)
-
-csv :: String
-csv =
-  """created,flag,foo,id
-2020-01-01T00:00:00.0Z,true,a,1
-2024-02-02T08:00:00.0Z,false,apple,2
-1970-01-01T00:00:00.0Z,true,hello,3
-"""
-
-dt :: String -> DateTime
-dt = toDateTimeLossy <<< unsafePartial fromJust <<< fromRFC3339String <<< wrap
-
-spec :: Spec Unit
-spec =
-  describe "Pipes.CSV" do
-    it "stringify" do
-      let
-        objs =
-          [ { id: 1, foo: "a", flag: true, created: dt "2020-01-01T00:00:00Z" }
-          , { id: 2, foo: "apple", flag: false, created: dt "2024-02-02T08:00:00Z" }
-          , { id: 3, foo: "hello", flag: true, created: dt "1970-01-01T00:00:00Z" }
-          ]
-
-      csv' <- map fold $ Pipes.Collect.toArray $ Pipes.Stream.withEOS (Pipes.Construct.eachArray objs) >-/-> Pipes.CSV.stringify >-> Pipes.Stream.unEOS
-      csv' `shouldEqual` csv
-    describe "parse" do
-      it "parses csv" do
-        rows <- map Array.fromFoldable
-          $ Pipes.toListM
-          $ Pipes.Stream.withEOS (yield csv)
-              >-> Pipes.Stream.inEOS (Pipes.Buffer.fromString UTF8)
-              >-/-> Pipes.CSV.parse
-              >-> Pipes.Stream.unEOS
-
-        rows `shouldEqual`
-          [ { id: 1, foo: "a", flag: true, created: dt "2020-01-01T00:00:00Z" }
-          , { id: 2, foo: "apple", flag: false, created: dt "2024-02-02T08:00:00Z" }
-          , { id: 3, foo: "hello", flag: true, created: dt "1970-01-01T00:00:00Z" }
-          ]
-      before
-        (do
-          nums <- liftEffect $ randomSample' 100000 (chooseInt 0 9)
-          let
-            chars = [ "i","d","\n" ] <> join ((\n -> [show n, "\n"]) <$> nums)
-          bufs <- Pipes.Collect.toArray
-            $ Pipes.Stream.withEOS (Pipes.Construct.eachArray chars)
-             >-> Pipes.Util.chunked 1000
-             >-> Pipes.Stream.inEOS (Pipes.map fold >-> Pipes.Buffer.fromString UTF8)
-             >-> Pipes.Stream.unEOS
-          pure $ nums /\ bufs
-        )
-        $ it "parses large csv" \(nums /\ bufs) -> do
-          rows <-
-            Pipes.Collect.toArray
-              $ Pipes.Stream.withEOS (Pipes.Construct.eachArray bufs)
-                  >-/-> Pipes.CSV.parse @(id :: Int)
-                  >-> Pipes.Stream.unEOS
-
-          rows `shouldEqual` ((\id -> { id }) <$> nums)