go fmt

* Add encoding/decoding HTTP msgs to Java middleware example.

* Add clojure middleware example.

.gitignore

@@ -15,3 +15,5 @@
 gor
 
 *.mprof
+
+*.pcap

.gitmodules

@@ -0,0 +1,15 @@
+[submodule "vendor/github.com/mattbaird/elastigo"]
+	path = vendor/github.com/mattbaird/elastigo
+	url = https://github.com/mattbaird/elastigo
+[submodule "vendor/github.com/araddon/gou"]
+	path = vendor/github.com/araddon/gou
+	url = https://github.com/araddon/gou
+[submodule "vendor/github.com/bitly/go-hostpool"]
+	path = vendor/github.com/bitly/go-hostpool
+	url = https://github.com/bitly/go-hostpool
+[submodule "vendor/github.com/bmizerany/assert"]
+	path = vendor/github.com/bmizerany/assert
+	url = https://github.com/bmizerany/assert
+[submodule "vendor/github.com/google/gopacket"]
+	path = vendor/github.com/google/gopacket
+	url = https://github.com/google/gopacket

COMM-LICENSE

@@ -2,7 +2,7 @@ END-USER LICENSE AGREEMENT
 
 ------------------------------------------------------------------------------
 
-IMPORTANT: THIS SOFTWARE END-USER LICENSE AGREEMENT ("EULA") IS A LEGAL AGREEMENT (“Agreement”) BETWEEN YOU (THE CUSTOMER, EITHER AS AN INDIVIDUAL OR, IF PURCHASED OR OTHERWISE ACQUIRED BY OR FOR AN ENTITY, AS AN ENTITY) AND REPLAY SOFTWARE. READ IT CAREFULLY BEFORE COMPLETING THE INSTALLATION PROCESS AND USING GOR PRO AND RELATED SOFTWARE COMPONENTS (“SOFTWARE”). IT PROVIDES A LICENSE TO USE THE SOFTWARE AND CONTAINS WARRANTY INFORMATION AND LIABILITY DISCLAIMERS. BY INSTALLING AND USING THE SOFTWARE, YOU ARE CONFIRMING YOUR ACCEPTANCE OF THE SOFTWARE AND AGREEING TO BECOME BOUND BY THE TERMS OF THIS AGREEMENT.
+IMPORTANT: THIS SOFTWARE END-USER LICENSE AGREEMENT ("EULA") IS A LEGAL AGREEMENT (“Agreement”) BETWEEN YOU (THE CUSTOMER, EITHER AS AN INDIVIDUAL OR, IF PURCHASED OR OTHERWISE ACQUIRED BY OR FOR AN ENTITY, AS AN ENTITY) AND GoReplay LLC. READ IT CAREFULLY BEFORE COMPLETING THE INSTALLATION PROCESS AND USING GOREPLAY PRO AND RELATED SOFTWARE COMPONENTS (“SOFTWARE”). IT PROVIDES A LICENSE TO USE THE SOFTWARE AND CONTAINS WARRANTY INFORMATION AND LIABILITY DISCLAIMERS. BY INSTALLING AND USING THE SOFTWARE, YOU ARE CONFIRMING YOUR ACCEPTANCE OF THE SOFTWARE AND AGREEING TO BECOME BOUND BY THE TERMS OF THIS AGREEMENT.
 
 ------------------------------------------------------------------------------
 
@@ -12,21 +12,19 @@ In order to use the Software under this Agreement, you must receive a “Source
 
 1.1 General Use. This Agreement grants you a non-exclusive, non-transferable, limited license to the use rights for the Software, without the right to grant sublicenses, subject to the terms and conditions in this Agreement. The Software is licensed, not sold.
 
-1.2 Unlimited Organization License. If you purchased an Organization License (included with the Gor Pro Software), you may install the Software on an unlimited number of Hosts. “Host” means any physical or virtual machine which is controlled by you. You may also run an unlimited number of Workers. “Worker” means a thread within a Gor server process which executes jobs. You may concurrently run the software on an unlimited number of Hosts, with each host running an unlimited number of Workers.
+1.2 Unlimited Organization License. If you purchased an Organization License (included with the GoReplay Pro Software), you may install the Software on an unlimited number of Hosts. “Host” means any physical or virtual machine which is controlled by you. You may concurrently run the software on an unlimited number of Hosts.
 
-1.3 Limited Organization License. If you purchased an Organization License (included with the Gor Enterprise Software), you may install the Software on an unlimited number of Hosts. “Host” means any physical or virtual machine which is controlled by you. The aggregate number of Workers run by the hosts must not exceed the maximum number of Workers authorized at the time of purchase. “Worker” means a running Gor instance which intercept or replay traffic. In order to run additional Workers, you must purchase an additional allowance from Replay Software.
+1.3 Appliance License. If you purchased an Appliance License, you may distribute the Software in any applications, frameworks, or elements (collectively referred to as an “Application” or “Applications”) that you develop using the Software in accordance with this EULA, provided that such distribution does not violate the restrictions set forth in section 3 of this EULA. You must not remove, obscure or interfere with any copyright, acknowledgment, attribution, trademark, warning or disclaimer statement affixed to, incorporated in or otherwise applied in connection with the Software. You are required to ensure that the Software is not reused by or with any applications other than those with which you distribute it as permitted herein. For example, if You install the Software on a customer’s server, that customer is not permitted to use the Software independently of your Application. You must inform GoReplay LLC of your knowledge of any infringing use of the Software by any of your customers. You are liable for compliance by those third parties with the terms and conditions of this EULA. You will not owe GoReplay LLC any royalties for your distribution of the Software in accordance with this EULA.
 
-1.4 Appliance License. If you purchased an Appliance License, you may distribute the Software in any applications, frameworks, or elements (collectively referred to as an “Application” or “Applications”) that you develop using the Software in accordance with this EULA, provided that such distribution does not violate the restrictions set forth in section 3 of this EULA. You must not remove, obscure or interfere with any copyright, acknowledgment, attribution, trademark, warning or disclaimer statement affixed to, incorporated in or otherwise applied in connection with the Software. You are required to ensure that the Software is not reused by or with any applications other than those with which you distribute it as permitted herein. For example, if You install the Software on a customer’s server, that customer is not permitted to use the Software independently of your Application. You must inform Replay Software of your knowledge of any infringing use of the Software by any of your customers. You are liable for compliance by those third parties with the terms and conditions of this EULA. You will not owe Replay Software any royalties for your distribution of the Software in accordance with this EULA.
+1.4 Archive Copies. You are entitled to make a reasonable amount of copies of the Software for archival purposes. Each copy must reproduce all copyright and other proprietary rights notices on or in the Software Product.
 
-1.5 Archive Copies. You are entitled to make a reasonable amount of copies of the Software for archival purposes. Each copy must reproduce all copyright and other proprietary rights notices on or in the Software Product.
+1.5 Electronic Delivery. All Software and license documentation shall be delivered by electronic means unless otherwise specified on the applicable invoice or at the time of purchase. Software shall be deemed delivered when it is made available for download by you (“Delivery”).        
 
-1.6 Electronic Delivery. All Software and license documentation shall be delivered by electronic means unless otherwise specified on the applicable invoice or at the time of purchase. Software shall be deemed delivered when it is made available for download by you (“Delivery”).        
-
-2. Modifications. Replay Software shall provide you with source code so that you can create Modifications of the original software. “Modification” means: (a) any addition to or deletion from the contents of a file included in the original Software or previous Modifications created by You, or (b) any new file that contains any part of the original Software or previous Modifications. While you retain all rights to any original work authored by you as part of the Modifications, We continue to own all copyright and other intellectual property rights in the Software.
+2. Modifications. GoReplay LLC shall provide you with source code so that you can create Modifications of the original software. “Modification” means: (a) any addition to or deletion from the contents of a file included in the original Software or previous Modifications created by You, or (b) any new file that contains any part of the original Software or previous Modifications. While you retain all rights to any original work authored by you as part of the Modifications, We continue to own all copyright and other intellectual property rights in the Software.
 
 3. Restricted Uses. 
 
-3.1 You shall not (and shall not allow any third party to): (a) decompile, disassemble, or otherwise reverse engineer the Software or attempt to reconstruct or discover any source code, underlying ideas, algorithms, file formats or programming interfaces of the Software by any means whatsoever (except and only to the extent that applicable law prohibits or restricts reverse engineering restrictions); (b) distribute, sell, sublicense, rent, lease or use the Software for time sharing, hosting, service provider or like purposes, except as expressly permitted under this Agreement; (c) redistribute the Software or Modifications other than by including the Software or a portion thereof within your own product, which must have substantially different functionality than the Software or Modifications and must not allow any third party to use the Software or Modifications, or any portions thereof, for software development or application development purposes; (d) redistribute the Software as part of a product, "appliance" or "virtual server"; (e) redistribute the Software on any server which is not directly under your control; (f) remove any product identification, proprietary, copyright or other notices contained in the Software; (g) modify any part of the Software, create a derivative work of any part of the Software (except as permitted in Section 4), or incorporate the Software, except to the extent expressly authorized in writing by Replay Software; (h) publicly disseminate performance information or analysis (including, without limitation, benchmarks) from any source relating to the Software; (i) utilize any equipment, device, software, or other means designed to circumvent or remove any form of Source URL or copy protection used by Replay Software in connection with the Software, or use the Software together with any authorization code, Source URL, serial number, or other copy protection device not supplied by Replay Software; (j) use the Software to develop a product which is competitive with any Replay Software product offerings; or (k) use unauthorized Source URLS or keycode(s) or distribute or publish Source URLs or keycode(s), except as may be expressly permitted by Replay Software in writing. If your unique Source URL is ever published, Replay Software reserves the right to terminate your access without notice.
+3.1 You shall not (and shall not allow any third party to): (a) decompile, disassemble, or otherwise reverse engineer the Software or attempt to reconstruct or discover any source code, underlying ideas, algorithms, file formats or programming interfaces of the Software by any means whatsoever (except and only to the extent that applicable law prohibits or restricts reverse engineering restrictions); (b) distribute, sell, sublicense, rent, lease or use the Software for time sharing, hosting, service provider or like purposes, except as expressly permitted under this Agreement; (c) redistribute the Software or Modifications other than by including the Software or a portion thereof within your own product, which must have substantially different functionality than the Software or Modifications and must not allow any third party to use the Software or Modifications, or any portions thereof, for software development or application development purposes; (d) redistribute the Software as part of a product, "appliance" or "virtual server"; (e) redistribute the Software on any server which is not directly under your control; (f) remove any product identification, proprietary, copyright or other notices contained in the Software; (g) modify any part of the Software, create a derivative work of any part of the Software (except as permitted in Section 4), or incorporate the Software, except to the extent expressly authorized in writing by GoReplay LLC; (h) publicly disseminate performance information or analysis (including, without limitation, benchmarks) from any source relating to the Software; (i) utilize any equipment, device, software, or other means designed to circumvent or remove any form of Source URL or copy protection used by GoReplay LLC in connection with the Software, or use the Software together with any authorization code, Source URL, serial number, or other copy protection device not supplied by GoReplay LLC; (j) use the Software to develop a product which is competitive with any GoReplay LLC product offerings; or (k) use unauthorized Source URLS or keycode(s) or distribute or publish Source URLs or keycode(s), except as may be expressly permitted by GoReplay LLC in writing. If your unique Source URL is ever published, GoReplay LLC reserves the right to terminate your access without notice.
 
 3.2 UNDER NO CIRCUMSTANCES MAY YOU USE THE SOFTWARE AS PART OF A PRODUCT OR SERVICE THAT PROVIDES SIMILAR FUNCTIONALITY TO THE SOFTWARE ITSELF.
 
@@ -34,34 +32,34 @@ The Open Source version of the Software (“LGPL Version”) is licensed
 under the terms of the GNU Lesser General Public License versions 3.0
 (“LGPL”) and not under this EULA. 
 
-4. Ownership. Notwithstanding anything to the contrary contained herein, except for the limited license rights expressly provided herein, Replay Software and its suppliers have and will retain all rights, title and interest (including, without limitation, all patent, copyright, trademark, trade secret and other intellectual property rights) in and to the Software and all copies, modifications and derivative works thereof (including any changes which incorporate any of your ideas, feedback or suggestions). You acknowledge that you are obtaining only a limited license right to the Software, and that irrespective of any use of the words “purchase”, “sale” or like terms hereunder no ownership rights are being conveyed to you under this Agreement or otherwise. 
+4. Ownership. Notwithstanding anything to the contrary contained herein, except for the limited license rights expressly provided herein, GoReplay LLC and its suppliers have and will retain all rights, title and interest (including, without limitation, all patent, copyright, trademark, trade secret and other intellectual property rights) in and to the Software and all copies, modifications and derivative works thereof (including any changes which incorporate any of your ideas, feedback or suggestions). You acknowledge that you are obtaining only a limited license right to the Software, and that irrespective of any use of the words “purchase”, “sale” or like terms hereunder no ownership rights are being conveyed to you under this Agreement or otherwise. 
 
-5. Fees and Payment. The Software license fees will be due and payable in full as set forth in the applicable invoice or at the time of purchase. If the Software does not function properly within two weeks of purchase, please contact us within those two weeks for a refund. You shall be responsible for all taxes, withholdings, duties and levies arising from the order (excluding taxes based on the net income of Replay Software). 
+5. Fees and Payment. The Software license fees will be due and payable in full as set forth in the applicable invoice or at the time of purchase. If the Software does not function properly within two weeks of purchase, please contact us within those two weeks for a refund. You shall be responsible for all taxes, withholdings, duties and levies arising from the order (excluding taxes based on the net income of GoReplay LLC). 
 
-6. Support, Maintenance and Services. Subject to the terms and conditions of this Agreement, as set forth in your invoice, and as set forth on the Gor Pro support page (https://github.com/buger/gor/wiki/Pro-Support), support and maintenance services may be included with the purchase of your license subscription.
+6. Support, Maintenance and Services. Subject to the terms and conditions of this Agreement, as set forth in your invoice, and as set forth on the GoReplay Pro support page (https://github.com/buger/gor/wiki/Pro-Support), support and maintenance services may be included with the purchase of your license subscription.
 
 7. Term of Agreement. 
 
-7.1 Term. This Agreement is effective as of the Delivery of the Software and expires at such time as all license and service subscriptions hereunder have expired in accordance with their own terms (the “Term”). For clarification, the term of your license under this Agreement may be perpetual, limited for Evaluation Version, or designated as a fixed-term license in the Invoice, and shall be specified at your time of purchase. Either party may terminate this Agreement (including all related Invoices) if the other party: (a) fails to cure any material breach of this Agreement within thirty (30) days after written notice of such breach, provided that Replay Software may terminate this Agreement immediately upon any breach of Section 3 or if you exceed any other restrictions contained in Section 1, unless otherwise specified in this agreement; (b) ceases operation without a successor; or (c) seeks protection under any bankruptcy, receivership, trust deed, creditors arrangement, composition or comparable proceeding, or if any such proceeding is instituted against such party (and not dismissed within sixty (60) days)). Termination is not an exclusive remedy and the exercise by either party of any remedy under this Agreement will be without prejudice to any other remedies it may have under this Agreement, by law, or otherwise.        
+7.1 Term. This Agreement is effective as of the Delivery of the Software and expires at such time as all license and service subscriptions hereunder have expired in accordance with their own terms (the “Term”). For clarification, the term of your license under this Agreement may be perpetual, limited for Evaluation Version, or designated as a fixed-term license in the Invoice, and shall be specified at your time of purchase. Either party may terminate this Agreement (including all related Invoices) if the other party: (a) fails to cure any material breach of this Agreement within thirty (30) days after written notice of such breach, provided that GoReplay LLC may terminate this Agreement immediately upon any breach of Section 3 or if you exceed any other restrictions contained in Section 1, unless otherwise specified in this agreement; (b) ceases operation without a successor; or (c) seeks protection under any bankruptcy, receivership, trust deed, creditors arrangement, composition or comparable proceeding, or if any such proceeding is instituted against such party (and not dismissed within sixty (60) days)). Termination is not an exclusive remedy and the exercise by either party of any remedy under this Agreement will be without prejudice to any other remedies it may have under this Agreement, by law, or otherwise.        
 
 7.2 Termination. Upon any termination of this Agreement, you shall cease any and all use of any Software and destroy all copies thereof. 
 
-7.3 Expiration of License. Upon the expiration of any term under this Agreement, (a) all Software updates and services pursuant to the license shall cease, (b) you may only continue to run existing installations of the Software, (c) you may not install the Software on any additional Hosts, and (d) any new installation of the Software shall require the purchase of a new license subscription from Replay Software.
+7.3 Expiration of License. Upon the expiration of any term under this Agreement, (a) all Software updates and services pursuant to the license shall cease, (b) you may only continue to run existing installations of the Software, (c) you may not install the Software on any additional Hosts, and (d) any new installation of the Software shall require the purchase of a new license subscription from GoReplay LLC.
 
-8. Disclaimer of Warranties. The Software is provided "as is," with all faults, defects and errors, and without warranty of any kind. Replay Software does not warrant that the Software will be free of bugs, errors, viruses or other defects, and Replay Software shall have no liability of any kind for the use of or inability to use the Software, the Software content or any associated service, and you acknowledge that it is not technically practicable for Replay Software to do so. 
-To the maximum extent permitted by applicable law, Replay Software disclaims all warranties, express, implied, arising by law or otherwise, regarding the Software, the Software content and their respective performance or suitability for your intended use, including without limitation any implied warranty of merchantability, fitness for a particular purpose.
+8. Disclaimer of Warranties. The Software is provided "as is," with all faults, defects and errors, and without warranty of any kind. GoReplay LLC does not warrant that the Software will be free of bugs, errors, viruses or other defects, and GoReplay LLC shall have no liability of any kind for the use of or inability to use the Software, the Software content or any associated service, and you acknowledge that it is not technically practicable for GoReplay LLC to do so. 
+To the maximum extent permitted by applicable law, GoReplay LLC disclaims all warranties, express, implied, arising by law or otherwise, regarding the Software, the Software content and their respective performance or suitability for your intended use, including without limitation any implied warranty of merchantability, fitness for a particular purpose.
 
 9. Limitation of Liability. 
 
-In no event will Replay Software be liable for any direct, indirect, consequential, incidental, special, exemplary, or punitive damages or liabilities whatsoever arising from or relating to the Software, the Software content or this Agreement, whether based on contract, tort (including negligence), strict liability or other theory, even if Replay Software has been advised of the possibility of such damages.
+In no event will GoReplay LLC be liable for any direct, indirect, consequential, incidental, special, exemplary, or punitive damages or liabilities whatsoever arising from or relating to the Software, the Software content or this Agreement, whether based on contract, tort (including negligence), strict liability or other theory, even if GoReplay LLC has been advised of the possibility of such damages.
 
-In no event will Replay Software' liability exceed the Software license price as indicated in the invoice. The existence of more than one claim will not enlarge or extend this limit.
+In no event will GoReplay LLC liability exceed the Software license price as indicated in the invoice. The existence of more than one claim will not enlarge or extend this limit.
 
-10. Remedies. Your exclusive remedy and Replay Software’ entire liability for breach of this Agreement shall be limited, at Replay Software’ sole and exclusive discretion, to (a) replacement of any defective software or documentation; or (b) refund of the license fee paid to Replay Software, payable in accordance with Replay Software' refund policy.
+10. Remedies. Your exclusive remedy and GoReplay LLC’ entire liability for breach of this Agreement shall be limited, at GoReplay LLC’ sole and exclusive discretion, to (a) replacement of any defective software or documentation; or (b) refund of the license fee paid to GoReplay LLC, payable in accordance with GoReplay LLC' refund policy.
 
 11. Acknowledgements.
 
-11.1 Consent to the Use of Data. You agree that Replay Software and its affiliates may collect and use technical information gathered as part of the product support services. Replay Software may use this information solely to improve products and services and will not disclose this information in a form that personally identifies you.
+11.1 Consent to the Use of Data. You agree that GoReplay LLC and its affiliates may collect and use technical information gathered as part of the product support services. GoReplay LLC may use this information solely to improve products and services and will not disclose this information in a form that personally identifies you.
 
 11.2 Verification. We or a certified auditor acting on our behalf, may, upon its reasonable request and at its expense, audit you with respect to the use of the Software. Such audit may be conducted by mail, electronic means or through an in-person visit to your place of business. Any such in-person audit shall be conducted during regular business hours at your facilities and shall not unreasonably interfere with your business activities. We shall not remove, copy, or redistribute any electronic material during the course of an audit. If an audit reveals that you are using the Software in a way that is in material violation of the terms of the EULA, then you shall pay our reasonable costs of conducting the audit. In the case of a material violation, you agree to pay Us any amounts owing that are attributable to the unauthorized use. In the alternative, We reserve the right, at our sole option, to terminate the licenses for the Software.
 
@@ -74,13 +72,13 @@ In no event will Replay Software' liability exceed the Software license price as
 
 13.1 Entire Agreement. This Agreement sets forth our entire agreement with respect to the Software and the subject matter hereof and supersedes all prior and contemporaneous understandings and agreements whether written or oral.
 
-13.2 Amendment. Replay Software reserves the right, in its sole discretion, to amend this Agreement from time. Amendments to this Agreement can be located at: https://github.com/buger/gor/blob/master/COMM-LICENSE.
+13.2 Amendment. GoReplay LLC reserves the right, in its sole discretion, to amend this Agreement from time. Amendments to this Agreement can be located at: https://github.com/buger/gor/blob/master/COMM-LICENSE.
 
-13.3 Assignment. You may not assign this Agreement or any of its rights under this Agreement without the prior written consent of Replay Software and any attempted assignment without such consent shall be void.
+13.3 Assignment. You may not assign this Agreement or any of its rights under this Agreement without the prior written consent of GoReplay LLC and any attempted assignment without such consent shall be void.
 
 13.4 Export Compliance. You agree to comply with all applicable laws and regulations, including laws, regulations, orders or other restrictions on export, re-export or redistribution of software.
 
-13.5 Indemnification. You agree to defend, indemnify, and hold harmless Replay Software from and against any lawsuits, claims, losses, damages, fines and expenses (including attorneys' fees and costs) arising out of your use of the Software or breach of this Agreement.
+13.5 Indemnification. You agree to defend, indemnify, and hold harmless GoReplay LLC from and against any lawsuits, claims, losses, damages, fines and expenses (including attorneys' fees and costs) arising out of your use of the Software or breach of this Agreement.
 
 13.6 Governing Law. This Agreement is governed by the laws of the State of Oregon and the United States without regard to conflicts of laws provisions thereof, and without regard to the United Nations Convention on the International Sale of Goods or the Uniform Computer Information Transactions Act, as currently enacted by any jurisdiction or as may be codified or amended from time to time by any jurisdiction. The jurisdiction and venue for actions related to the subject matter hereof shall be the state of Oregon and United States federal courts located in Portland, Oregon, and both parties hereby submit to the personal jurisdiction of such courts. 
 
@@ -92,4 +90,4 @@ In no event will Replay Software' liability exceed the Software license price as
 
 13.10 Headings. The headings of sections and paragraphs of this Agreement are for convenience of reference only and are not intended to restrict, affect or be of any weight in the interpretation or construction of the provisions of such sections or paragraphs.
 
-14. Contact Information. If you have any questions about this EULA, or if you want to contact Replay Software for any reason, please direct correspondence to info@gortool.com
+14. Contact Information. If you have any questions about this EULA, or if you want to contact GoReplay LLC for any reason, please direct correspondence to info@gortool.com

LICENSE.txt

@@ -1,7 +1,19 @@
-Gor is an Open Source project licensed under the terms of
+GoReplay is an Open Source project licensed under the terms of
 the LGPLv3 license.  Please see <http://www.gnu.org/licenses/lgpl-3.0.html>
 for license text.
 
-Gor Pro has a commercial-friendly license allowing private forks
-and modifications of Gor.  Please see http://gortool.com/#pro for
+As a special exception to the GNU Lesser General Public License version 3
+("LGPL3"), the copyright holders of this Library give you permission to
+convey to a third party a Combined Work that links statically or dynamically
+to this Library without providing any Minimal Corresponding Source or
+Minimal Application Code as set out in 4d or providing the installation
+information set out in section 4e, provided that you comply with the other
+provisions of LGPL3 and provided that you meet, for the Application the
+terms and conditions of the license(s) which apply to the Application.
+
+TLDR: You are free to use Gor subpackages like `byteutils` or `proto` in your commercial projects.
+
+
+GoReplay Pro has a commercial-friendly license allowing private forks
+and modifications of GoReplay.  Please see https://goreplay.org/pro.html for
 more detail.  You can find the commercial license terms in COMM-LICENSE.

Makefile

@@ -1,4 +1,4 @@
-SOURCE = emitter.go gor.go gor_stat.go input_dummy.go input_file.go input_raw.go input_tcp.go limiter.go output_dummy.go output_file.go input_http.go output_http.go output_tcp.go plugins.go settings.go test_input.go elasticsearch.go http_modifier.go http_modifier_settings.go http_client.go middleware.go protocol.go output_file_settings.go
+SOURCE = emitter.go gor.go gor_stat.go input_dummy.go input_file.go input_raw.go input_tcp.go limiter.go output_dummy.go output_null.go output_file.go input_http.go output_http.go output_tcp.go plugins.go settings.go test_input.go elasticsearch.go http_modifier.go http_modifier_settings.go http_client.go middleware.go protocol.go output_file_settings.go
 SOURCE_PATH = /go/src/github.com/buger/gor/
 PORT = 8000
 FADDR = :8000
@@ -13,10 +13,10 @@ FADDR = ":8000"
 release: release-x64 release-mac
 
 release-x64:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i gor go build $(LDFLAGS) && tar -czf gor_$(VERSION)_x64.tar.gz gor && rm gor
+	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=amd64  -i gor go build -tags netgo $(LDFLAGS) && tar -czf gor_$(VERSION)_x64.tar.gz gor && rm gor
 
 release-x86:
-	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i gor go build $(LDFLAGS) && tar -czf gor_$(VERSION)_x86.tar.gz gor && rm gor
+	docker run -v `pwd`:$(SOURCE_PATH) -t --env GOOS=linux --env GOARCH=386 -i gor go build -tags netgo $(LDFLAGS) && tar -czf gor_$(VERSION)_x86.tar.gz gor && rm gor
 
 release-mac:
 	go build $(MAC_LDFLAGS) && tar -czf gor_$(VERSION)_mac.tar.gz gor && rm gor
@@ -62,7 +62,7 @@ profile_test:
 
 # Used mainly for debugging, because docker container do not have access to parent machine ports
 run:
-	$(RUN) go run $(LDFLAGS) $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw-track-response --input-raw 127.0.0.1:9000 --input-http 127.0.0.1:9000 --verbose --debug --middleware "./examples/middleware/echo.sh" --output-file requests.gor
+	$(RUN) go run $(LDFLAGS) $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw-track-response --input-raw 127.0.0.1:9000 --verbose --debug --middleware "./examples/middleware/echo.sh" --output-file requests.gor
 
 run-2:
 	sudo -E go run $(SOURCE) --input-dummy="" --output-tcp localhost:27001 --verbose --debug
@@ -76,6 +76,9 @@ run-arg:
 file-server:
 	go run $(SOURCE) file-server $(FADDR)
 
+readpcap:
+	go run $(SOURCE) --input-raw $(FILE) --input-raw-engine pcap_file --output-null
+
 record:
 	$(RUN) go run $(SOURCE) --input-dummy=0 --output-file=requests.gor --verbose --debug
 

README.md

@@ -2,17 +2,24 @@
 
 ![Go Replay](http://i.imgur.com/ZG2ki5n.png)
 
+## https://goreplay.org/
+
 ## About
 
-Gor is an open-source tool for capturing and replaying live HTTP traffic into a test environment in order to continuously test your system with real data. It can be used to increase confidence in code deployments, configuration changes and infrastructure changes.
+GoReplay is the simplest and safest way to test your app using real traffic before you put it into production.
+
+As your application grows, the effort required to test it also grows exponentially. GoReplay offers you the simple idea of reusing your existing traffic for testing, which makes it incredibly powerful. Our state of art technique allows to analyze and record your application traffic without affecting it. This eliminates the risks that come with putting a third party component in the critical path. 
+
+GoReplay increases your confidence in code deployments, configuration changes and infrastructure changes. Did we mention that no coding is required?
+
 
-Now you can test your code on real user sessions in an automated and repeatable fashion.
-**No more falling down in production!**
 
 Here is basic workflow: The listener server catches http traffic and sends it to the replay server or saves to file. The replay server forwards traffic to a given address.
 
 ![Diagram](http://i.imgur.com/9mqj2SK.png)
 
+Check [latest documentation](http://github.com/buger/gor/wiki).
+
 ## Installation
 Download latest binary from https://github.com/buger/gor/releases or [compile by yourself](https://github.com/buger/gor/wiki/Compilation).
 
@@ -21,14 +28,15 @@ Download latest binary from https://github.com/buger/gor/releases or [compile by
 The most basic setup will be `sudo ./gor --input-raw :8000 --output-stdout` which acts like tcpdump.
 If you already have test environment you can start replaying: `sudo ./gor --input-raw :8000 --output-http http://staging.env`.
 
-See the our wiki and especially [Getting started](https://github.com/buger/gor/wiki/Getting-Started) wiki page for more info. 
+See the our [documentation](https://github.com/buger/gor/wiki/) and [Getting started](https://github.com/buger/gor/wiki/Getting-Started) page for more info. 
+## Newsletter
+Subscribe to our [newsletter](https://www.getdrip.com/forms/89690474/submissions/new) to stay informed about the latest features and changes to Gor project.
 
 
 ## Want to Upgrade?
 
-I also sell Gor Pro, extensions to Gor which provide more features, a commercial-friendly license and allow you to support high quality open source development all at the same time. Please see the Gor [homepage](https://gortool.com/) for more detail.
+We have created a [GoReplay PRO](https://goreplay.org/pro.html) extension which provides additional features such as support for binary protocols like Thrift or ProtocolBuffers, saving and replaying from cloud storage, TCP sessions replication, etc. The PRO version also includes a commercial-friendly license, dedicated support, and it also allows you to support high-quality open source development. 
 
-Subscribe to the [quarterly newsletter](https://tinyletter.com/gor) to stay informed about the latest features and changes to Gor and its bigger siblings.
 
 ## Problems?
 If you have a problem, please review the [FAQ](https://github.com/buger/gor/wiki/FAQ) and [Troubleshooting](https://github.com/buger/gor/wiki/Troubleshooting) wiki pages. Searching the [issues](https://github.com/buger/gor/issues) for your problem is also a good idea.
@@ -36,14 +44,6 @@ If you have a problem, please review the [FAQ](https://github.com/buger/gor/wiki
 All bug-reports and suggestions should go though Github Issues or our [Google Group](https://groups.google.com/forum/#!forum/gor-users) (you can just send email to gor-users@googlegroups.com).
 If you have a private question feel free to send email to support@gortool.com.
 
-Useful resources:
-
-* Product documentation is in the [wiki](http://github.com/buger/gor/wiki).
-* Release announcements are made to the [@buger](http://twitter.com/buger) Twitter account and our [newsleter](https://tinyletter.com/gor)
-
-
-If you need commercial support read more about Pro and Enterprise versions at our site [https://gortool.com/](https://gortool.com/)
-
 
 ## Contributing
 
@@ -66,7 +66,7 @@ If you need commercial support read more about Pro and Enterprise versions at ou
 * [Granify](http://granify.com) - AI backed SaaS solution that enables online retailers to maximise their sales
 * And many more!
 
-If you are using Gor we are happy add you to the list and share your story, just write to: hello@gortool.com
+If you are using Gor we are happy add you to the list and share your story, just write to: hello@goreplay.org
 
 ## Author
 

elasticsearch.go

@@ -2,9 +2,8 @@ package main
 
 import (
 	"encoding/json"
-	"github.com/buger/elastigo/api"
-	"github.com/buger/elastigo/core"
 	"github.com/buger/gor/proto"
+	"github.com/mattbaird/elastigo/lib"
 	"log"
 	"regexp"
 	"time"
@@ -19,33 +18,34 @@ func (e *ESUriErorr) Error() string {
 type ESPlugin struct {
 	Active  bool
 	ApiPort string
+	eConn   *elastigo.Conn
 	Host    string
 	Index   string
-	indexor *core.BulkIndexer
+	indexor *elastigo.BulkIndexer
 	done    chan bool
 }
 
 type ESRequestResponse struct {
-	ReqUrl               []byte `json:"Req_URL"`
-	ReqMethod            []byte `json:"Req_Method"`
-	ReqUserAgent         []byte `json:"Req_User-Agent"`
-	ReqAcceptLanguage    []byte `json:"Req_Accept-Language,omitempty"`
-	ReqAccept            []byte `json:"Req_Accept,omitempty"`
-	ReqAcceptEncoding    []byte `json:"Req_Accept-Encoding,omitempty"`
-	ReqIfModifiedSince   []byte `json:"Req_If-Modified-Since,omitempty"`
-	ReqConnection        []byte `json:"Req_Connection,omitempty"`
-	ReqCookies           []byte `json:"Req_Cookies,omitempty"`
-	RespStatus           []byte `json:"Resp_Status"`
-	RespStatusCode       []byte `json:"Resp_Status-Code"`
-	RespProto            []byte `json:"Resp_Proto,omitempty"`
-	RespContentLength    []byte `json:"Resp_Content-Length,omitempty"`
-	RespContentType      []byte `json:"Resp_Content-Type,omitempty"`
-	RespTransferEncoding []byte `json:"Resp_Transfer-Encoding,omitempty"`
-	RespContentEncoding  []byte `json:"Resp_Content-Encoding,omitempty"`
-	RespExpires          []byte `json:"Resp_Expires,omitempty"`
-	RespCacheControl     []byte `json:"Resp_Cache-Control,omitempty"`
-	RespVary             []byte `json:"Resp_Vary,omitempty"`
-	RespSetCookie        []byte `json:"Resp_Set-Cookie,omitempty"`
+	ReqURL               string `json:"Req_URL"`
+	ReqMethod            string `json:"Req_Method"`
+	ReqUserAgent         string `json:"Req_User-Agent"`
+	ReqAcceptLanguage    string `json:"Req_Accept-Language,omitempty"`
+	ReqAccept            string `json:"Req_Accept,omitempty"`
+	ReqAcceptEncoding    string `json:"Req_Accept-Encoding,omitempty"`
+	ReqIfModifiedSince   string `json:"Req_If-Modified-Since,omitempty"`
+	ReqConnection        string `json:"Req_Connection,omitempty"`
+	ReqCookies           string `json:"Req_Cookies,omitempty"`
+	RespStatus           string `json:"Resp_Status"`
+	RespStatusCode       string `json:"Resp_Status-Code"`
+	RespProto            string `json:"Resp_Proto,omitempty"`
+	RespContentLength    string `json:"Resp_Content-Length,omitempty"`
+	RespContentType      string `json:"Resp_Content-Type,omitempty"`
+	RespTransferEncoding string `json:"Resp_Transfer-Encoding,omitempty"`
+	RespContentEncoding  string `json:"Resp_Content-Encoding,omitempty"`
+	RespExpires          string `json:"Resp_Expires,omitempty"`
+	RespCacheControl     string `json:"Resp_Cache-Control,omitempty"`
+	RespVary             string `json:"Resp_Vary,omitempty"`
+	RespSetCookie        string `json:"Resp_Set-Cookie,omitempty"`
 	Rtt                  int64  `json:"RTT"`
 	Timestamp            time.Time
 }
@@ -76,24 +76,24 @@ func (p *ESPlugin) Init(URI string) {
 	if err != nil {
 		log.Fatal("Can't initialize ElasticSearch plugin.", err)
 	}
+	p.eConn = elastigo.NewConn()
+	p.eConn.SetPort(p.ApiPort)
+	p.eConn.SetHosts([]string{p.Host})
 
-	api.Domain = p.Host
-	api.Port = p.ApiPort
-
-	p.indexor = core.NewBulkIndexerErrors(50, 60)
+	p.indexor = p.eConn.NewBulkIndexerErrors(50, 60)
 	p.done = make(chan bool)
-	p.indexor.Run(p.done)
+	p.indexor.Start()
 
 	// Only start the ErrorHandler goroutine when in verbose mode
 	// no need to burn ressources otherwise
-	// go p.ErrorHandler()
+	go p.ErrorHandler()
 
 	log.Println("Initialized Elasticsearch Plugin")
 	return
 }
 
 func (p *ESPlugin) IndexerShutdown() {
-	p.done <- true
+	p.indexor.Stop()
 	return
 }
 
@@ -118,28 +118,29 @@ func (p *ESPlugin) ResponseAnalyze(req, resp []byte, start, stop time.Time) {
 	}
 	t := time.Now()
 	rtt := p.RttDurationToMs(stop.Sub(start))
+	req = payloadBody(req)
 
 	esResp := ESRequestResponse{
-		ReqUrl:               proto.Path(req),
-		ReqMethod:            proto.Method(req),
-		ReqUserAgent:         proto.Header(req, []byte("User-Agent")),
-		ReqAcceptLanguage:    proto.Header(req, []byte("Accept-Language")),
-		ReqAccept:            proto.Header(req, []byte("Accept")),
-		ReqAcceptEncoding:    proto.Header(req, []byte("Accept-Encoding")),
-		ReqIfModifiedSince:   proto.Header(req, []byte("If-Modified-Since")),
-		ReqConnection:        proto.Header(req, []byte("Connection")),
-		ReqCookies:           proto.Header(req, []byte("Cookie")),
-		RespStatus:           proto.Status(resp),
-		RespStatusCode:       proto.Status(resp),
-		RespProto:            proto.Method(resp),
-		RespContentLength:    proto.Header(resp, []byte("Content-Length")),
-		RespContentType:      proto.Header(resp, []byte("Content-Type")),
-		RespTransferEncoding: proto.Header(resp, []byte("Transfer-Encoding")),
-		RespContentEncoding:  proto.Header(resp, []byte("Content-Encoding")),
-		RespExpires:          proto.Header(resp, []byte("Expires")),
-		RespCacheControl:     proto.Header(resp, []byte("Cache-Control")),
-		RespVary:             proto.Header(resp, []byte("Vary")),
-		RespSetCookie:        proto.Header(resp, []byte("Set-Cookie")),
+		ReqURL:               string(proto.Path(req)),
+		ReqMethod:            string(proto.Method(req)),
+		ReqUserAgent:         string(proto.Header(req, []byte("User-Agent"))),
+		ReqAcceptLanguage:    string(proto.Header(req, []byte("Accept-Language"))),
+		ReqAccept:            string(proto.Header(req, []byte("Accept"))),
+		ReqAcceptEncoding:    string(proto.Header(req, []byte("Accept-Encoding"))),
+		ReqIfModifiedSince:   string(proto.Header(req, []byte("If-Modified-Since"))),
+		ReqConnection:        string(proto.Header(req, []byte("Connection"))),
+		ReqCookies:           string(proto.Header(req, []byte("Cookie"))),
+		RespStatus:           string(proto.Status(resp)),
+		RespStatusCode:       string(proto.Status(resp)),
+		RespProto:            string(proto.Method(resp)),
+		RespContentLength:    string(proto.Header(resp, []byte("Content-Length"))),
+		RespContentType:      string(proto.Header(resp, []byte("Content-Type"))),
+		RespTransferEncoding: string(proto.Header(resp, []byte("Transfer-Encoding"))),
+		RespContentEncoding:  string(proto.Header(resp, []byte("Content-Encoding"))),
+		RespExpires:          string(proto.Header(resp, []byte("Expires"))),
+		RespCacheControl:     string(proto.Header(resp, []byte("Cache-Control"))),
+		RespVary:             string(proto.Header(resp, []byte("Vary"))),
+		RespSetCookie:        string(proto.Header(resp, []byte("Set-Cookie"))),
 		Rtt:                  rtt,
 		Timestamp:            t,
 	}
@@ -147,7 +148,7 @@ func (p *ESPlugin) ResponseAnalyze(req, resp []byte, start, stop time.Time) {
 	if err != nil {
 		log.Println(err)
 	} else {
-		p.indexor.Index(p.Index, "RequestResponse", "", "", &t, j, true)
+		p.indexor.Index(p.Index, "RequestResponse", "", "", "", &t, j)
 	}
 	return
 }

emitter.go

@@ -32,13 +32,7 @@ func Start(stop chan int) {
 	for {
 		select {
 		case <-stop:
-			pluginMu.Lock()
-			for _, p := range Plugins.All {
-				if cp, ok := p.(io.Closer); ok {
-					cp.Close()
-				}
-			}
-			pluginMu.Unlock()
+			finalize()
 			return
 		case <-time.After(100 * time.Millisecond):
 		}

examples/middleware/echo.clj

@@ -0,0 +1,53 @@
+(ns echo.core
+  (:gen-class)
+  (:require [clojure.string :as cs]
+            [clojure.java.io :as io])
+  (:import org.apache.commons.codec.binary.Hex
+           java.io.BufferedReader
+           java.io.IOException
+           java.io.InputStreamReader))
+
+
+(defn transform-http-msg
+  "Function that transforms/filters the incoming HTTP messages."
+  [headers body]
+  ;; do actual transformations here
+  [headers body])
+
+
+(defn decode-hex-string
+  "Decode an Hex-encoded string."
+  [s]
+  (String. (Hex/decodeHex (.toCharArray s))))
+
+
+(defn encode-hex-string
+  "Encode a string to a hex-encoded string."
+  [^String s]
+  (String. (Hex/encodeHex (.getBytes s))))
+
+
+(defn -main
+  [& args]
+  (let [br (BufferedReader. (InputStreamReader. System/in))]
+    (try
+      (loop [hex-line (.readLine br)]
+        (let [decoded-req (decode-hex-string hex-line)
+
+              ;; empty line separates headers from body
+              http-request (partition-by empty? (cs/split-lines decoded-req))
+              headers (first http-request)
+
+              ;; HTTP messages can contain no body:
+              body (when (= 3 (count http-request)) (last http-request))
+              [new-headers new-body] (transform-http-msg headers body)]
+
+          (println (encode-hex-string (str (cs/join "\n" headers)
+                                           (when body
+                                             (str "\n\n"
+                                                  (cs/join "\n" body)))))))
+        (when-let [line (.readLine br)]
+          (recur line)))
+      (catch IOException e nil))))
+
+

examples/middleware/echo.java

@@ -2,8 +2,25 @@ import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
 
-public class echo {
-    public static void main(String[] args) {
+import org.apache.commons.codec.DecoderException;
+import org.apache.commons.codec.binary.Hex;
+
+
+public class Echo {
+        public static String decodeHexString(String s) throws DecoderException {
+            return new String(Hex.decodeHex(s.toCharArray()));
+        }
+
+        public static String encodeHexString(String s) {
+            return new String(Hex.encodeHex(s.getBytes()));
+        }
+
+        public static String transformHTTPMessage(String req) {
+            // do actual transformations here
+            return req;
+        }
+
+    public static void main(String[] args) throws DecoderException {
         if(args != null){
             for(String arg : args){
                 System.out.println(arg);
@@ -17,11 +34,15 @@ public class echo {
 
         try {
             while ((line = stdin.readLine()) != null) {
+                String decodedLine = decodeHexString(line);
 
-                System.out.println(line);
+                String transformedLine = transformHTTPMessage(decodedLine);
+
+                String encodedLine = encodeHexString(transformedLine);
+                System.out.println(encodedLine);
 
             }
         } catch (IOException e) {
         }
     }
-}
+}

gor.go

@@ -78,17 +78,31 @@ func main() {
 	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
 	go func() {
 		<-c
-
-		for _, p := range Plugins.All {
-			if cp, ok := p.(io.Closer); ok {
-				cp.Close()
-			}
-		}
-
+		finalize()
 		os.Exit(1)
 	}()
 
-	Start(nil)
+	if Settings.exitAfter > 0 {
+		log.Println("Running gor for a duration of", Settings.exitAfter)
+		closeCh := make(chan int)
+
+		time.AfterFunc(Settings.exitAfter, func() {
+			log.Println("Stopping gor after", Settings.exitAfter)
+			close(closeCh)
+		})
+
+		Start(closeCh)
+	} else {
+		Start(nil)
+	}
+}
+
+func finalize() {
+	for _, p := range Plugins.All {
+		if cp, ok := p.(io.Closer); ok {
+			cp.Close()
+		}
+	}
 }
 
 func profileCPU(cpuprofile string) {

http_client.go

@@ -3,6 +3,7 @@ package main
 import (
 	"bytes"
 	"crypto/tls"
+	"encoding/base64"
 	"io"
 	"log"
 	"net"
@@ -44,6 +45,7 @@ type HTTPClient struct {
 	baseURL        string
 	scheme         string
 	host           string
+	auth           string
 	conn           net.Conn
 	respBuf        []byte
 	config         *HTTPClientConfig
@@ -79,6 +81,10 @@ func NewHTTPClient(baseURL string, config *HTTPClientConfig) *HTTPClient {
 	client.respBuf = make([]byte, config.ResponseBufferSize)
 	client.config = config
 
+	if u.User != nil {
+		client.auth = "Basic " + base64.StdEncoding.EncodeToString([]byte(u.User.String()))
+	}
+
 	return client
 }
 
@@ -164,6 +170,10 @@ func (c *HTTPClient) Send(data []byte) (response []byte, err error) {
 		data = proto.SetHost(data, []byte(c.baseURL), []byte(c.host))
 	}
 
+	if c.auth != "" {
+		data = proto.SetHeader(data, []byte("Authorization"), []byte(c.auth))
+	}
+
 	if c.config.Debug {
 		Debug("[HTTPClient] Sending:", string(data))
 	}

http_client_test.go

@@ -11,6 +11,7 @@ import (
 	"net/http/httptest"
 	"net/http/httputil"
 	_ "reflect"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -87,6 +88,37 @@ func TestHTTPClientSend(t *testing.T) {
 	wg.Wait()
 }
 
+func TestHTTPClientResonseByClose(t *testing.T) {
+	wg := new(sync.WaitGroup)
+
+	payload := []byte("GET / HTTP/1.1\r\n\r\n")
+	ln, _ := net.Listen("tcp", ":0")
+	go func() {
+		for {
+			conn, _ := ln.Accept()
+			buf := make([]byte, 4096)
+			conn.Read(buf)
+
+			conn.Write([]byte("HTTP/1.1 200 OK\r\n\r\n"))
+			conn.Write([]byte("ab"))
+			conn.Close()
+
+			wg.Done()
+		}
+	}()
+
+	client := NewHTTPClient(ln.Addr().String(), &HTTPClientConfig{Debug: true})
+
+	wg.Add(1)
+	resp, _ := client.Send(payload)
+
+	if !bytes.Equal(resp, []byte("HTTP/1.1 200 OK\r\n\r\nab")) {
+		t.Error("Should return valid response", string(resp))
+	}
+
+	wg.Wait()
+}
+
 // https://github.com/buger/gor/issues/184
 func TestHTTPClientResponseBuffer(t *testing.T) {
 	testCases := []struct {
@@ -325,6 +357,45 @@ func TestHTTPClientRedirectLimit(t *testing.T) {
 	wg.Wait()
 }
 
+func TestHTTPClientBasicAuth(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	wg.Add(2)
+
+	GETPayload := []byte("GET / HTTP/1.1\r\n\r\n")
+
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		user, pass, _ := r.BasicAuth()
+
+		if user != "user" || pass != "pass" {
+			http.Error(w, "Unauthorized.", 401)
+			wg.Done()
+			return
+		}
+
+		wg.Done()
+	}))
+	defer server.Close()
+
+	client := NewHTTPClient(server.URL, &HTTPClientConfig{Debug: false})
+	resp, _ := client.Send(GETPayload)
+	client.Disconnect()
+
+	if !bytes.Equal(proto.Status(resp), []byte("401")) {
+		t.Error("Should return unauthorized error", string(resp))
+	}
+
+	authUrl := strings.Replace(server.URL, "http://", "http://user:pass@", -1)
+	client = NewHTTPClient(authUrl, &HTTPClientConfig{Debug: false})
+	resp, _ = client.Send(GETPayload)
+	client.Disconnect()
+
+	if !bytes.Equal(proto.Status(resp), []byte("200")) {
+		t.Error("Should return proper response", string(resp))
+	}
+
+	wg.Wait()
+}
+
 func TestHTTPClientHandleHTTP10(t *testing.T) {
 	wg := new(sync.WaitGroup)
 

input_dummy.go

@@ -34,10 +34,10 @@ func (i *DummyInput) emit() {
 		select {
 		case <-ticker.C:
 			uuid := uuid()
-			reqh := payloadHeader(RequestPayload, uuid, time.Now().UnixNano())
+			reqh := payloadHeader(RequestPayload, uuid, time.Now().UnixNano(), -1)
 			i.data <- append(reqh, []byte("GET / HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\n\r\n")...)
 
-			resh := payloadHeader(ResponsePayload, uuid, 1)
+			resh := payloadHeader(ResponsePayload, uuid, time.Now().UnixNano()+1, 1)
 			i.data <- append(resh, []byte("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")...)
 		}
 	}

input_file.go

@@ -9,31 +9,114 @@ import (
 	"log"
 	"os"
 	"path/filepath"
-	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
+type fileInputReader struct {
+	reader    *bufio.Reader
+	data      []byte
+	file      *os.File
+	timestamp int64
+}
+
+func (f *fileInputReader) parseNext() error {
+	payloadSeparatorAsBytes := []byte(payloadSeparator)
+	var buffer bytes.Buffer
+
+	for {
+		line, err := f.reader.ReadBytes('\n')
+
+		if err != nil {
+			if err != io.EOF {
+				log.Println(err)
+				return err
+			}
+
+			if err == io.EOF {
+				f.file.Close()
+				f.file = nil
+				return err
+			}
+		}
+
+		if bytes.Equal(payloadSeparatorAsBytes[1:], line) {
+			asBytes := buffer.Bytes()
+			meta := payloadMeta(asBytes)
+
+			f.timestamp, _ = strconv.ParseInt(string(meta[2]), 10, 64)
+			f.data = asBytes[:len(asBytes)-1]
+
+			return nil
+		}
+
+		buffer.Write(line)
+	}
+
+	return nil
+}
+
+func (f *fileInputReader) ReadPayload() []byte {
+	defer f.parseNext()
+
+	return f.data
+}
+func (f *fileInputReader) Close() error {
+	if f.file != nil {
+		f.file.Close()
+	}
+
+	return nil
+}
+
+func NewFileInputReader(path string) *fileInputReader {
+	file, err := os.Open(path)
+
+	if err != nil {
+		log.Println(err)
+		return nil
+	}
+
+	r := &fileInputReader{file: file}
+	if strings.HasSuffix(path, ".gz") {
+		gzReader, err := gzip.NewReader(file)
+		if err != nil {
+			log.Println(err)
+			return nil
+		}
+		r.reader = bufio.NewReader(gzReader)
+	} else {
+		r.reader = bufio.NewReader(file)
+	}
+
+	r.parseNext()
+
+	return r
+}
+
 // FileInput can read requests generated by FileOutput
 type FileInput struct {
-	data          chan []byte
-	path          string
-	currentFile   *os.File
-	currentReader *bufio.Reader
-	speedFactor   float64
-	loop          bool
+	mu          sync.Mutex
+	data        chan []byte
+	exit        chan bool
+	path        string
+	readers     []*fileInputReader
+	speedFactor float64
+	loop        bool
 }
 
 // NewFileInput constructor for FileInput. Accepts file path as argument.
 func NewFileInput(path string, loop bool) (i *FileInput) {
 	i = new(FileInput)
-	i.data = make(chan []byte)
+	i.data = make(chan []byte, 1000)
+	i.exit = make(chan bool, 1)
 	i.path = path
 	i.speedFactor = 1
 	i.loop = loop
 
-	if err := i.updateFile(); err != nil {
+	if err := i.init(); err != nil {
 		return
 	}
 
@@ -48,9 +131,10 @@ func (_ *NextFileNotFound) Error() string {
 	return "There is no new files"
 }
 
-// path can be a pattern
-// It sort paths lexicographically and tries to choose next one
-func (i *FileInput) updateFile() (err error) {
+func (i *FileInput) init() (err error) {
+	defer i.mu.Unlock()
+	i.mu.Lock()
+
 	var matches []string
 
 	if matches, err = filepath.Glob(i.path); err != nil {
@@ -63,40 +147,10 @@ func (i *FileInput) updateFile() (err error) {
 		return errors.New("No matching files")
 	}
 
-	sort.Sort(sortByFileIndex(matches))
+	i.readers = make([]*fileInputReader, len(matches))
 
-	// Just pick first file, if there is many, and we are just started
-	if i.currentFile == nil {
-		if i.currentFile, err = os.Open(matches[0]); err != nil {
-			log.Println("Can't read file ", matches[0], err)
-			return
-		}
-	} else {
-		found := false
-		for idx, p := range matches {
-			if p == i.currentFile.Name() && idx != len(matches)-1 {
-				if i.currentFile, err = os.Open(matches[idx+1]); err != nil {
-					log.Println("Can't read file ", matches[idx+1], err)
-					return
-				}
-
-				found = true
-			}
-		}
-
-		if !found {
-			return new(NextFileNotFound)
-		}
-	}
-
-	if strings.HasSuffix(i.currentFile.Name(), ".gz") {
-		gzReader, err := gzip.NewReader(i.currentFile)
-		if err != nil {
-			log.Fatal(err)
-		}
-		i.currentReader = bufio.NewReader(gzReader)
-	} else {
-		i.currentReader = bufio.NewReader(i.currentFile)
+	for idx, p := range matches {
+		i.readers[idx] = NewFileInputReader(p)
 	}
 
 	return nil
@@ -113,83 +167,72 @@ func (i *FileInput) String() string {
 	return "File input: " + i.path
 }
 
-func (i *FileInput) emit() {
-	var lastTime int64
+// Find reader with smallest timestamp e.g next payload in row
+func (i *FileInput) nextReader() (next *fileInputReader) {
+	for _, r := range i.readers {
+		if r == nil || r.file == nil {
+			continue
+		}
 
-	payloadSeparatorAsBytes := []byte(payloadSeparator)
-
-	var buffer bytes.Buffer
-
-	if i.currentReader == nil {
-		return
+		if next == nil || r.timestamp < next.timestamp {
+			next = r
+			continue
+		}
 	}
 
+	return
+}
+
+func (i *FileInput) emit() {
+	var lastTime int64 = -1
+
 	for {
-		line, err := i.currentReader.ReadBytes('\n')
+		select {
+		case <-i.exit:
+			return
+		default:
+		}
 
-		if err != nil {
-			if err != io.EOF {
-				log.Fatal(err)
-			}
-
-			// If our path pattern match multiple files, try to find them
-			if err == io.EOF {
-				if e := i.updateFile(); e != nil {
-					if _, ok := e.(*NextFileNotFound); ok && i.loop {
-						// Start from the first file
-						i.Close()
-						i.currentFile = nil
-						i.currentReader = nil
-						lastTime = 0
-						i.updateFile()
-
-						continue
-					} else {
-						break
-					}
-				}
+		reader := i.nextReader()
 
+		if reader == nil {
+			if i.loop {
+				i.init()
+				lastTime = -1
 				continue
+			} else {
+				break
 			}
 		}
 
-		if bytes.Equal(payloadSeparatorAsBytes[1:], line) {
-			asBytes := buffer.Bytes()
-			buffer.Reset()
+		if lastTime != -1 {
+			diff := reader.timestamp - lastTime
+			lastTime = reader.timestamp
 
-			meta := payloadMeta(asBytes)
-
-			if len(meta) > 2 && meta[0][0] == RequestPayload {
-				ts, _ := strconv.ParseInt(string(meta[2]), 10, 64)
-
-				if lastTime != 0 {
-					timeDiff := ts - lastTime
-
-					if i.speedFactor != 1 {
-						timeDiff = int64(float64(timeDiff) / i.speedFactor)
-					}
-
-					time.Sleep(time.Duration(timeDiff))
-				}
-
-				lastTime = ts
+			if i.speedFactor != 1 {
+				diff = int64(float64(diff) / i.speedFactor)
 			}
 
-			// Bytes() returns only pointer, so to remove data-race copy the data to an array
-			newBuf := make([]byte, len(asBytes)-1)
-			copy(newBuf, asBytes)
-
-			i.data <- newBuf
+			time.Sleep(time.Duration(diff))
 		} else {
-			buffer.Write(line)
+			lastTime = reader.timestamp
 		}
 
+		i.data <- reader.ReadPayload()
 	}
 
 	log.Printf("FileInput: end of file '%s'\n", i.path)
 }
 
 func (i *FileInput) Close() error {
-	i.currentFile.Close()
+	defer i.mu.Unlock()
+	i.mu.Lock()
+
+	i.exit <- true
+
+	for _, r := range i.readers {
+		r.Close()
+	}
+
 	return nil
 }

input_file_test.go

@@ -97,35 +97,109 @@ func TestInputFileWithGETAndPOST(t *testing.T) {
 
 }
 
-func TestInputFileMultipleFiles(t *testing.T) {
+func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) {
 	rnd := rand.Int63()
 
 	file1, _ := os.OpenFile(fmt.Sprintf("/tmp/%d_0", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
 	file1.Write([]byte("1 1 1\ntest1"))
 	file1.Write([]byte(payloadSeparator))
-	file1.Write([]byte("1 1 2\ntest2"))
+	file1.Write([]byte("1 1 3\ntest2"))
 	file1.Write([]byte(payloadSeparator))
 	file1.Close()
 
 	file2, _ := os.OpenFile(fmt.Sprintf("/tmp/%d_1", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
-	file2.Write([]byte("1 1 3\ntest3"))
+	file2.Write([]byte("1 1 2\ntest3"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Write([]byte("1 1 4\ntest4"))
 	file2.Write([]byte(payloadSeparator))
 	file2.Close()
 
 	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
 	buf := make([]byte, 1000)
-	n, _ := input.Read(buf)
-	if buf[10] != '1' {
-		t.Error("Shound emit requests in right order", string(buf[:n]))
-	}
-	input.Read(buf)
-	if buf[10] != '2' {
-		t.Error("Shound emit requests in right order", string(buf[:n]))
+
+	for i := '1'; i <= '4'; i++ {
+		n, _ := input.Read(buf)
+		if buf[4] != byte(i) {
+			t.Error("Should emit requests in right order", string(buf[:n]))
+		}
 	}
 
-	input.Read(buf)
-	if buf[10] != '3' {
-		t.Error("Shound emit requests from second file", string(buf[:n]))
+	os.Remove(file1.Name())
+	os.Remove(file2.Name())
+}
+
+func TestInputFileRequestsWithLatency(t *testing.T) {
+	rnd := rand.Int63()
+
+	file, _ := os.OpenFile(fmt.Sprintf("/tmp/%d", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
+	defer file.Close()
+
+	file.Write([]byte("1 1 100000000\nrequest1"))
+	file.Write([]byte(payloadSeparator))
+	file.Write([]byte("1 2 150000000\nrequest2"))
+	file.Write([]byte(payloadSeparator))
+	file.Write([]byte("1 3 250000000\nrequest3"))
+	file.Write([]byte(payloadSeparator))
+
+	input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false)
+	buf := make([]byte, 1000)
+
+	start := time.Now().UnixNano()
+	for i := 0; i < 3; i++ {
+		input.Read(buf)
+	}
+	end := time.Now().UnixNano()
+
+	var expectedLatency int64 = 250000000 - 100000000
+	realLatency := end - start
+	if realLatency < expectedLatency {
+		t.Errorf("Should emit requests respecting latency. Expected: %v, real: %v", expectedLatency, realLatency)
+	}
+
+	if realLatency > expectedLatency+10000000 {
+		t.Errorf("Should emit requests respecting latency. Expected: %v, real: %v", expectedLatency, realLatency)
+
+	}
+}
+
+func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) {
+	rnd := rand.Int63()
+
+	file1, _ := os.OpenFile(fmt.Sprintf("/tmp/%d_0", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
+	file1.Write([]byte("1 1 1\nrequest1"))
+	file1.Write([]byte(payloadSeparator))
+	file1.Write([]byte("2 1 1\nresponse1"))
+	file1.Write([]byte(payloadSeparator))
+	file1.Write([]byte("1 2 3\nrequest2"))
+	file1.Write([]byte(payloadSeparator))
+	file1.Write([]byte("2 2 3\nresponse2"))
+	file1.Write([]byte(payloadSeparator))
+	file1.Close()
+
+	file2, _ := os.OpenFile(fmt.Sprintf("/tmp/%d_1", rnd), os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0660)
+	file2.Write([]byte("1 3 2\nrequest3"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Write([]byte("2 3 2\nresponse3"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Write([]byte("1 4 4\nrequest4"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Write([]byte("2 4 4\nresponse4"))
+	file2.Write([]byte(payloadSeparator))
+	file2.Close()
+
+	input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false)
+	buf := make([]byte, 1000)
+
+	for i := '1'; i <= '4'; i++ {
+		n, _ := input.Read(buf)
+		if buf[0] != '1' && buf[4] != byte(i) {
+			t.Error("Shound emit requests in right order", string(buf[:n]))
+		}
+
+		n, _ = input.Read(buf)
+		if buf[0] != '2' && buf[4] != byte(i) {
+			t.Error("Shound emit responses in right order", string(buf[:n]))
+		}
 	}
 
 	os.Remove(file1.Name())
@@ -149,8 +223,8 @@ func TestInputFileLoop(t *testing.T) {
 	for i := 0; i < 1000; i++ {
 		input.Read(buf)
 	}
-	input.Close()
 
+	input.Close()
 	os.Remove(file.Name())
 }
 
@@ -221,11 +295,9 @@ func (expectedCaptureFile *CaptureFile) PayloadsEqual(other [][]byte) bool {
 	}
 
 	for i, payload := range other {
-
 		if !bytes.Equal(expectedCaptureFile.data[i], payload) {
 			return false
 		}
-
 	}
 
 	return true

input_http.go

@@ -29,7 +29,7 @@ func NewHTTPInput(address string) (i *HTTPInput) {
 func (i *HTTPInput) Read(data []byte) (int, error) {
 	buf := <-i.data
 
-	header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano())
+	header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1)
 
 	copy(data[0:len(header)], header)
 	copy(data[len(header):], buf)

input_raw.go

@@ -24,6 +24,7 @@ type RAWInput struct {
 const (
 	EngineRawSocket = 1 << iota
 	EnginePcap
+	EnginePcapFile
 )
 
 // NewRAWInput constructor for RAWInput. Accepts address with port as argument.
@@ -50,12 +51,12 @@ func (i *RAWInput) Read(data []byte) (int, error) {
 	var header []byte
 
 	if msg.IsIncoming {
-		header = payloadHeader(RequestPayload, msg.UUID(), msg.Start.UnixNano())
+		header = payloadHeader(RequestPayload, msg.UUID(), msg.Start.UnixNano(), -1)
 		if len(i.realIPHeader) > 0 {
 			buf = proto.SetHeader(buf, i.realIPHeader, []byte(msg.IP().String()))
 		}
 	} else {
-		header = payloadHeader(ResponsePayload, msg.UUID(), msg.End.UnixNano()-msg.AssocMessage.Start.UnixNano())
+		header = payloadHeader(ResponsePayload, msg.UUID(), msg.AssocMessage.Start.UnixNano(), msg.End.UnixNano()-msg.AssocMessage.Start.UnixNano())
 	}
 
 	copy(data[0:len(header)], header)
@@ -69,6 +70,12 @@ func (i *RAWInput) listen(address string) {
 
 	host, port, err := net.SplitHostPort(address)
 
+	if i.engine == EnginePcapFile {
+		host = address
+		port = "1"
+		err = nil
+	}
+
 	if err != nil {
 		log.Fatal("input-raw: error while parsing address", err)
 	}

input_raw_test.go

@@ -79,6 +79,55 @@ func TestRAWInputIPv4(t *testing.T) {
 	}
 
 	wg.Wait()
+
+	close(quit)
+}
+
+func TestRAWInputNoKeepAlive(t *testing.T) {
+	wg := new(sync.WaitGroup)
+	quit := make(chan int)
+
+	listener, err := net.Listen("tcp", ":0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	origin := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.Write([]byte("a"))
+			w.Write([]byte("b"))
+		}),
+		ReadTimeout:  10 * time.Second,
+		WriteTimeout: 10 * time.Second,
+	}
+	origin.SetKeepAlivesEnabled(false)
+	go origin.Serve(listener)
+	defer listener.Close()
+
+	originAddr := listener.Addr().String()
+
+	input := NewRAWInput(originAddr, EnginePcap, true, testRawExpire, "")
+	defer input.Close()
+
+	output := NewTestOutput(func(data []byte) {
+		wg.Done()
+	})
+
+	Plugins.Inputs = []io.Reader{input}
+	Plugins.Outputs = []io.Writer{output}
+
+	client := NewHTTPClient("http://"+listener.Addr().String(), &HTTPClientConfig{})
+
+	go Start(quit)
+
+	for i := 0; i < 100; i++ {
+		// request + response
+		wg.Add(2)
+		client.Get("/")
+		time.Sleep(2 * time.Millisecond)
+	}
+
+	wg.Wait()
+
 	close(quit)
 }
 

output_file.go

@@ -8,6 +8,7 @@ import (
 	"log"
 	"os"
 	"path/filepath"
+	"runtime/debug"
 	"sort"
 	"strconv"
 	"strings"
@@ -52,14 +53,6 @@ func NewFileOutput(pathTemplate string, config *FileOutputConfig) *FileOutput {
 	o.config = config
 	o.updateName()
 
-	// Force flushing every minute
-	go func() {
-		for {
-			time.Sleep(o.config.flushInterval)
-			o.flush()
-		}
-	}()
-
 	go func() {
 		for {
 			time.Sleep(time.Second)
@@ -208,6 +201,13 @@ func (o *FileOutput) Write(data []byte) (n int, err error) {
 }
 
 func (o *FileOutput) flush() {
+	// Don't exit on panic
+	defer func() {
+		if r := recover(); r != nil {
+			log.Println("PANIC while file flush: ", r, o, string(debug.Stack()))
+		}
+	}()
+
 	defer o.mu.Unlock()
 	o.mu.Lock()
 

output_http.go

@@ -14,6 +14,7 @@ type response struct {
 	payload       []byte
 	uuid          []byte
 	roundTripTime int64
+	startedAt     int64
 }
 
 // HTTPOutputConfig struct for holding http output configuration
@@ -178,7 +179,7 @@ func (o *HTTPOutput) Read(data []byte) (int, error) {
 
 	Debug("[OUTPUT-HTTP] Received response:", string(resp.payload))
 
-	header := payloadHeader(ReplayedResponsePayload, resp.uuid, resp.roundTripTime)
+	header := payloadHeader(ReplayedResponsePayload, resp.uuid, resp.roundTripTime, resp.startedAt)
 	copy(data[0:len(header)], header)
 	copy(data[len(header):], resp.payload)
 
@@ -206,7 +207,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) {
 	}
 
 	if o.config.TrackResponses {
-		o.responses <- response{resp, uuid, stop.UnixNano() - start.UnixNano()}
+		o.responses <- response{resp, uuid, start.UnixNano(), stop.UnixNano() - start.UnixNano()}
 	}
 
 	if o.elasticSearch != nil {

output_null.go

@@ -0,0 +1,18 @@
+package main
+
+// NullOutput used for debugging, prints nothing
+type NullOutput struct {
+}
+
+// NullOutput constructor for NullOutput
+func NewNullOutput() (o *NullOutput) {
+	return new(NullOutput)
+}
+
+func (o *NullOutput) Write(data []byte) (int, error) {
+	return len(data), nil
+}
+
+func (o *NullOutput) String() string {
+	return "Null Output"
+}

plugins.go

@@ -95,9 +95,15 @@ func InitPlugins() {
 		registerPlugin(NewDummyOutput)
 	}
 
+	if Settings.outputNull {
+		registerPlugin(NewNullOutput)
+	}
+
 	engine := EnginePcap
 	if Settings.inputRAWEngine == "raw_socket" {
 		engine = EngineRawSocket
+	} else if Settings.inputRAWEngine == "pcap_file" {
+		engine = EnginePcapFile
 	}
 
 	for _, options := range Settings.inputRAW {

proto/proto.go

@@ -133,7 +133,7 @@ func header(payload []byte, name []byte) (value []byte, headerStart, headerEnd,
 	valueStart = headerStart + len(name) + 1 // Skip ":" after header name
 	headerEnd = valueStart + bytes.IndexByte(payload[valueStart:], '\n')
 
-	for valueStart < headerEnd {          // Ignore empty space after ':'
+	for valueStart < headerEnd { // Ignore empty space after ':'
 		if payload[valueStart] == ' ' {
 			valueStart++
 		} else {
@@ -148,7 +148,7 @@ func header(payload []byte, name []byte) (value []byte, headerStart, headerEnd,
 	}
 
 	// ignore empty space at end of header value
-	for valueStart < valueEnd  {
+	for valueStart < valueEnd {
 		if payload[valueEnd-1] == ' ' {
 			valueEnd--
 		} else {
@@ -160,6 +160,127 @@ func header(payload []byte, name []byte) (value []byte, headerStart, headerEnd,
 	return
 }
 
+// Works only with ASCII
+func HeadersEqual(h1 []byte, h2 []byte) bool {
+	if len(h1) != len(h2) {
+		return false
+	}
+
+	for i, c1 := range h1 {
+		c2 := h2[i]
+
+		switch int(c1) - int(c2) {
+		case 0, 32, -32:
+		default:
+			return false
+		}
+	}
+
+	return true
+}
+
+// Parsing headers from multiple payloads
+func ParseHeaders(payloads [][]byte, cb func(header []byte, value []byte) bool) {
+
+	hS := [2]int{0, 0}
+	hE := [2]int{-1, -1}
+	vS := [2]int{-1, -1}
+	vE := [2]int{-1, -1}
+
+	i := 0
+	pIdx := 0
+	lineBreaks := 0
+
+	for {
+		if len(payloads)-1 < pIdx {
+			break
+		}
+
+		p := payloads[pIdx]
+
+		if len(p)-1 < i {
+			pIdx++
+			i = 0
+			continue
+		}
+
+		switch p[i] {
+		case '\r', '\n':
+			lineBreaks++
+
+			// End of headers
+			if lineBreaks == 4 {
+				return
+			}
+
+			if lineBreaks > 1 {
+				break
+			}
+
+			vE = [2]int{pIdx, i}
+
+			if vS[1] != -1 && vE[1] != -1 &&
+				hS[1] != -1 && hE[1] != -1 {
+
+				var header, value []byte
+
+				phS, phE, pvS, pvE := payloads[hS[0]], payloads[hE[0]], payloads[vS[0]], payloads[vE[0]]
+
+				// If in same payload
+				if hS[0] == hE[0] {
+					header = phS[hS[1]:hE[1]]
+				} else {
+					header = make([]byte, len(phS)-hS[1]+hE[1])
+					copy(header, phS[hS[1]:])
+					copy(header[len(phS)-hS[1]:], phE[:hE[1]])
+				}
+
+				if vS[0] == vE[0] {
+					value = pvS[vS[1]:vE[1]]
+				} else {
+					value = make([]byte, len(pvS)-vS[1]+vE[1])
+					copy(value, pvS[vS[1]:])
+					copy(value[len(pvS)-vS[1]:], pvE[:vE[1]])
+				}
+
+				if !cb(header, value) {
+					return
+				}
+			}
+
+			// Header found, reset values
+			vS = [2]int{-1, -1}
+			vE = [2]int{-1, -1}
+			hS = [2]int{-1, -1}
+			hE = [2]int{-1, -1}
+		case ':':
+			hE = [2]int{pIdx, i}
+		default:
+			lineBreaks = 0
+
+			if hS[1] == -1 {
+				hS = [2]int{pIdx, i}
+			} else {
+				if hE[1] == -1 {
+					break
+				}
+
+				if vS[1] == -1 {
+					if p[i] == ' ' {
+						break
+					}
+
+					vS = [2]int{pIdx, i}
+				}
+			}
+		}
+
+		i++
+	}
+
+	return
+}
+
 // Header returns header value, if header not found, value will be blank
 func Header(payload, name []byte) []byte {
 	val, _, _, _, _ := header(payload, name)
@@ -170,7 +291,7 @@ func Header(payload, name []byte) []byte {
 // SetHeader sets header value. If header not found it creates new one.
 // Returns modified request payload
 func SetHeader(payload, name, value []byte) []byte {
-	_, hs, _, vs, ve  := header(payload, name)
+	_, hs, _, vs, ve := header(payload, name)
 
 	if hs != -1 {
 		// If header found we just replace its value
@@ -196,12 +317,12 @@ func AddHeader(payload, name, value []byte) []byte {
 
 // DelHeader takes http payload and removes header name from headers section
 // Returns modified request payload
-func DeleteHeader(payload, name[]byte) []byte {
+func DeleteHeader(payload, name []byte) []byte {
 	_, hs, he, _, _ := header(payload, name)
 	if hs != -1 {
-		newHeader := make([]byte, len(payload) - (he - hs) - 1)
+		newHeader := make([]byte, len(payload)-(he-hs)-1)
 		copy(newHeader[:hs], payload[:hs])
-		copy(newHeader[hs:], payload[he + 1:])
+		copy(newHeader[hs:], payload[he+1:])
 		return newHeader
 	}
 	return payload
@@ -322,7 +443,7 @@ func Status(payload []byte) []byte {
 }
 
 var httpMethods []string = []string{
-	"GET ", "OPTI", "HEAD", "POST", "PUT ", "DELE", "TRAC", "CONN", /* custom methods */"BAN", "PURG",
+	"GET ", "OPTI", "HEAD", "POST", "PUT ", "DELE", "TRAC", "CONN" /* custom methods */, "BAN", "PURG",
 }
 
 func IsHTTPPayload(payload []byte) bool {

proto/proto_test.go

@@ -2,6 +2,7 @@ package proto
 
 import (
 	"bytes"
+	"reflect"
 	"testing"
 )
 
@@ -123,6 +124,45 @@ func TestDeleteHeader(t *testing.T) {
 	}
 }
 
+func TestParseHeaders(t *testing.T) {
+	payload := [][]byte{[]byte("POST /post HTTP/1.1\r\nContent-Length: 7\r\nHost: www.w3.or"), []byte("g\r\nUser-Ag"), []byte("ent:Chrome\r\n\r\n"), []byte("Fake-Header: asda")}
+
+	headers := make(map[string]string)
+
+	ParseHeaders(payload, func(header []byte, value []byte) bool {
+		headers[string(header)] = string(value)
+		return true
+	})
+
+	expected := map[string]string{
+		"Content-Length": "7",
+		"Host":           "www.w3.org",
+		"User-Agent":     "Chrome",
+	}
+	if !reflect.DeepEqual(headers, expected) {
+		t.Error("Headers do not properly parsed", headers)
+	}
+}
+
+func TestHeaderEquals(t *testing.T) {
+	tests := []struct {
+		h1     string
+		h2     string
+		equals bool
+	}{
+		{"Content-Length", "content-length", true},
+		{"content-length", "Content-Length", true},
+		{"content-Pength", "Content-Length", false},
+		{"Host", "Content-Length", false},
+	}
+
+	for _, tc := range tests {
+		if HeadersEqual([]byte(tc.h1), []byte(tc.h2)) != tc.equals {
+			t.Error(tc)
+		}
+	}
+}
+
 func TestPath(t *testing.T) {
 	var path, payload []byte
 

protocol.go

@@ -42,13 +42,24 @@ func payloadScanner(data []byte, atEOF bool) (advance int, token []byte, err err
 }
 
 // Timing is request start or round-trip time, depending on payloadType
-func payloadHeader(payloadType byte, uuid []byte, timing int64) (header []byte) {
-	sTime := strconv.FormatInt(timing, 10)
+func payloadHeader(payloadType byte, uuid []byte, timing int64, latency int64) (header []byte) {
+	var sTime, sLatency string
+
+	sTime = strconv.FormatInt(timing, 10)
+	if latency != -1 {
+		sLatency = strconv.FormatInt(latency, 10)
+	}
 
 	//Example:
 	//  3 f45590522cd1838b4a0d5c5aab80b77929dea3b3 1231\n
 	// `+ 1` indicates space characters or end of line
-	header = make([]byte, 1+1+len(uuid)+1+len(sTime)+1)
+	headerLen := 1 + 1 + len(uuid) + 1 + len(sTime) + 1
+
+	if latency != -1 {
+		headerLen += len(sLatency) + 1
+	}
+
+	header = make([]byte, headerLen)
 	header[0] = payloadType
 	header[1] = ' '
 	header[2+len(uuid)] = ' '
@@ -57,6 +68,11 @@ func payloadHeader(payloadType byte, uuid []byte, timing int64) (header []byte)
 	copy(header[2:], uuid)
 	copy(header[3+len(uuid):], sTime)
 
+	if latency != -1 {
+		header[3+len(uuid)+len(sTime)] = ' '
+		copy(header[4+len(uuid)+len(sTime):], sLatency)
+	}
+
 	return header
 }
 

raw_socket_listener/listener.go

@@ -16,6 +16,7 @@ import (
 	"bytes"
 	"encoding/binary"
 	"fmt"
+	"github.com/buger/gor/proto"
 	"github.com/google/gopacket"
 	"github.com/google/gopacket/layers"
 	"github.com/google/gopacket/pcap"
@@ -28,7 +29,6 @@ import (
 	"strings"
 	"sync"
 	"time"
-	"github.com/buger/gor/proto"
 )
 
 var _ = fmt.Println
@@ -81,6 +81,7 @@ type request struct {
 const (
 	EngineRawSocket = 1 << iota
 	EnginePcap
+	EnginePcapFile
 )
 
 // NewListener creates and initializes new Listener object
@@ -114,10 +115,10 @@ func NewListener(addr string, port string, engine int, trackResponse bool, expir
 	// Special case for testing
 	if l.port != 0 {
 		switch engine {
-		case EngineRawSocket:
-			go l.readRAWSocket()
 		case EnginePcap:
 			go l.readPcap()
+		case EnginePcapFile:
+			go l.readPcapFile()
 		default:
 			log.Fatal("Unknown traffic interception engine:", engine)
 		}
@@ -173,7 +174,14 @@ func (t *Listener) dispatchMessage(message *TCPMessage) {
 
 	t.deleteMessage(message)
 
-	// log.Println("Dispatching, message", message.Start.UnixNano(), message.Seq, message.Ack, string(message.Bytes()))
+	if !message.complete {
+		if !message.IsIncoming {
+			delete(t.respAliases, message.Ack)
+			delete(t.respWithoutReq, message.Ack)
+		}
+
+		return
+	}
 
 	if message.IsIncoming {
 		// If there were response before request
@@ -183,10 +191,10 @@ func (t *Listener) dispatchMessage(message *TCPMessage) {
 				if resp, rok := t.messages[respID]; rok {
 					// if resp.AssocMessage == nil {
 					// log.Println("FOUND RESPONSE")
-					resp.AssocMessage = message
-					message.AssocMessage = resp
+					resp.setAssocMessage(message)
+					message.setAssocMessage(resp)
 
-					if resp.IsFinished() {
+					if resp.complete {
 						defer t.dispatchMessage(resp)
 					}
 					// }
@@ -194,14 +202,14 @@ func (t *Listener) dispatchMessage(message *TCPMessage) {
 			}
 
 			if resp, ok := t.messages[message.ResponseID]; ok {
-				resp.AssocMessage = message
+				resp.setAssocMessage(message)
 			}
 		}
 	} else {
 		if message.AssocMessage == nil {
 			if responseRequest, ok := t.respAliases[message.Ack]; ok {
-				message.AssocMessage = responseRequest
-				responseRequest.AssocMessage = message
+				message.setAssocMessage(responseRequest)
+				responseRequest.setAssocMessage(message)
 			}
 		}
 
@@ -245,6 +253,28 @@ func (e *DeviceNotFoundError) Error() string {
 	return msg
 }
 
+func isLoopback(device pcap.Interface) bool {
+	if len(device.Addresses) == 0 {
+		return false
+	}
+
+	switch device.Addresses[0].IP.String() {
+	case "127.0.0.1", "::1":
+		return true
+	}
+
+	return false
+}
+
+func listenAllInterfaces(addr string) bool {
+	switch addr {
+	case "", "0.0.0.0", "[::]", "::":
+		return true
+	default:
+		return false
+	}
+}
+
 func findPcapDevices(addr string) (interfaces []pcap.Interface, err error) {
 	devices, err := pcap.FindAllDevs()
 	if err != nil {
@@ -252,7 +282,7 @@ func findPcapDevices(addr string) (interfaces []pcap.Interface, err error) {
 	}
 
 	for _, device := range devices {
-		if (addr == "" || addr == "0.0.0.0" || addr == "[::]" || addr == "::") && len(device.Addresses) > 0 {
+		if listenAllInterfaces(addr) && len(device.Addresses) > 0 || isLoopback(device) {
 			interfaces = append(interfaces, device)
 			continue
 		}
@@ -300,12 +330,26 @@ func (t *Listener) readPcap() {
 			t.pcapHandles = append(t.pcapHandles, handle)
 
 			var bpfDstHost, bpfSrcHost string
-			for i, addr := range device.Addresses {
-				bpfDstHost += "dst host " + addr.IP.String()
-				bpfSrcHost += "src host " + addr.IP.String()
-				if i != len(device.Addresses) - 1 {
-					bpfDstHost += " or "
-					bpfSrcHost += " or "
+			var loopback = isLoopback(device)
+
+			if loopback {
+				var allAddr []string
+				for _, dc := range devices {
+					for _, addr := range dc.Addresses {
+						allAddr = append(allAddr, "(dst host "+addr.IP.String()+" and src host "+addr.IP.String()+")")
+					}
+				}
+
+				bpfDstHost = strings.Join(allAddr, " or ")
+				bpfSrcHost = bpfDstHost
+			} else {
+				for i, addr := range device.Addresses {
+					bpfDstHost += "dst host " + addr.IP.String()
+					bpfSrcHost += "src host " + addr.IP.String()
+					if i != len(device.Addresses)-1 {
+						bpfDstHost += " or "
+						bpfSrcHost += " or "
+					}
 				}
 			}
 
@@ -330,9 +374,9 @@ func (t *Listener) readPcap() {
 
 			// Special case for tunnel interface https://github.com/google/gopacket/issues/99
 			if handle.LinkType() == 12 {
-			    decoder = layers.LayerTypeIPv4
+				decoder = layers.LayerTypeIPv4
 			} else {
-			    decoder = handle.LinkType()
+				decoder = handle.LinkType()
 			}
 
 			source := gopacket.NewPacketSource(handle, decoder)
@@ -355,28 +399,29 @@ func (t *Listener) readPcap() {
 				// We should remove network layer before parsing TCP/IP data
 				var of int
 				switch decoder {
-					case layers.LinkTypeEthernet:
-						of = 14
-					case layers.LinkTypePPP:
-						of = 1
-					case layers.LinkTypeFDDI:
-						of = 13
-					case layers.LinkTypeNull:
-						of = 4
-					case layers.LinkTypeLoop:
-						of = 4
-					case layers.LinkTypeRaw:
-						of = 0
-					case layers.LinkTypeLinuxSLL:
-						of = 16
-					default:
-						log.Println("Unknown packet layer", packet)
-						break
+				case layers.LinkTypeEthernet:
+					of = 14
+				case layers.LinkTypePPP:
+					of = 1
+				case layers.LinkTypeFDDI:
+					of = 13
+				case layers.LinkTypeNull:
+					of = 4
+				case layers.LinkTypeLoop:
+					of = 4
+				case layers.LinkTypeRaw:
+					of = 0
+				case layers.LinkTypeLinuxSLL:
+					of = 16
+				default:
+					log.Println("Unknown packet layer", packet)
+					break
 				}
 
 				data = packet.Data()[of:]
 
 				version := uint8(data[0]) >> 4
+				ipLength := int(binary.BigEndian.Uint16(data[2:4]))
 
 				if version == 4 {
 					ihl := uint8(data[0]) & 0x0F
@@ -388,6 +433,24 @@ func (t *Listener) readPcap() {
 
 					srcIP = data[12:16]
 					dstIP = data[16:20]
+
+					// Too small IP packet
+					if ipLength < 20 {
+						continue
+					}
+
+					// Invalid length
+					if int(ihl*4) > ipLength {
+						continue
+					}
+
+					if cmp := len(data) - ipLength; cmp > 0 {
+						data = data[:ipLength]
+					} else if cmp < 0 {
+						// Truncated packet
+						continue
+					}
+
 					data = data[ihl*4:]
 				} else {
 					// Truncated IP info
@@ -407,10 +470,11 @@ func (t *Listener) readPcap() {
 				}
 
 				dataOffset := (data[12] & 0xF0) >> 4
+				isFIN := data[13]&0x01 != 0
 
 				// We need only packets with data inside
 				// Check that the buffer is larger than the size of the TCP header
-				if len(data) > int(dataOffset*4) {
+				if len(data) > int(dataOffset*4) || isFIN {
 					if !bpfSupported {
 						destPort := binary.BigEndian.Uint16(data[2:4])
 						srcPort := binary.BigEndian.Uint16(data[0:2])
@@ -430,10 +494,26 @@ func (t *Listener) readPcap() {
 						}
 
 						addrMatched := false
-						for _, a := range device.Addresses {
-							if a.IP.Equal(net.IP(addrCheck)) {
-								addrMatched = true
-								break
+
+						if loopback {
+							for _, dc := range devices {
+								if addrMatched {
+									break
+								}
+								for _, a := range dc.Addresses {
+									if a.IP.Equal(net.IP(addrCheck)) {
+										addrMatched = true
+										break
+									}
+								}
+							}
+							addrMatched = true
+						} else {
+							for _, a := range device.Addresses {
+								if a.IP.Equal(net.IP(addrCheck)) {
+									addrMatched = true
+									break
+								}
 							}
 						}
 
@@ -456,6 +536,68 @@ func (t *Listener) readPcap() {
 	t.readyCh <- true
 }
 
+func (t *Listener) readPcapFile() {
+	if handle, err := pcap.OpenOffline(t.addr); err != nil {
+		log.Fatal(err)
+	} else {
+		t.readyCh <- true
+		packetSource := gopacket.NewPacketSource(handle, handle.LinkType())
+
+		for {
+			packet, err := packetSource.NextPacket()
+			if err == io.EOF {
+				break
+			} else if err != nil {
+				log.Println("Error:", err)
+				continue
+			}
+
+			var addr, data []byte
+
+			if tcpLayer := packet.Layer(layers.LayerTypeTCP); tcpLayer != nil {
+				tcp, _ := tcpLayer.(*layers.TCP)
+				data = append(tcp.LayerContents(), tcp.LayerPayload()...)
+
+				if tcp.SrcPort >= 32768 && tcp.SrcPort <= 61000 {
+					copy(data[0:2], []byte{0, 0})
+					copy(data[2:4], []byte{0, 1})
+				} else {
+					copy(data[0:2], []byte{0, 1})
+					copy(data[2:4], []byte{0, 0})
+				}
+			} else {
+				continue
+			}
+
+			if ipLayer := packet.Layer(layers.LayerTypeIPv4); ipLayer != nil {
+				ip, _ := ipLayer.(*layers.IPv4)
+				addr = ip.SrcIP
+			} else if ipLayer = packet.Layer(layers.LayerTypeIPv6); ipLayer != nil {
+				ip, _ := ipLayer.(*layers.IPv6)
+				addr = ip.SrcIP
+			} else {
+				// log.Println("Can't find IP layer", packet)
+				continue
+			}
+
+			dataOffset := (data[12] & 0xF0) >> 4
+			isFIN := data[13]&0x01 != 0
+
+			// We need only packets with data inside
+			// Check that the buffer is larger than the size of the TCP header
+			if len(data) <= int(dataOffset*4) && !isFIN {
+				continue
+			}
+
+			newBuf := make([]byte, len(data)+16)
+			copy(newBuf[:16], addr)
+			copy(newBuf[16:], data)
+
+			t.packetsChan <- newBuf
+		}
+	}
+}
+
 func (t *Listener) readRAWSocket() {
 	conn, e := net.ListenPacket("ip:tcp", t.addr)
 	t.conn = conn
@@ -527,8 +669,6 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 		}
 	}()
 
-	// log.Println("Processing packet:", packet.Ack, packet.Seq, packet.ID)
-
 	var message *TCPMessage
 
 	isIncoming := packet.DestPort == t.port
@@ -541,7 +681,7 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 				t.deleteMessage(m)
 
 				if m.AssocMessage != nil {
-					m.AssocMessage.AssocMessage = nil
+					m.setAssocMessage(nil)
 				}
 
 				for _, pkt := range m.packets {
@@ -557,6 +697,14 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 		packet.UpdateAck(parentAck)
 	}
 
+	if isIncoming && packet.IsFIN {
+		if ma, ok := t.respAliases[packet.Seq]; ok {
+			if ma.packets[0].SrcPort == packet.SrcPort {
+				packet.UpdateAck(ma.Ack)
+			}
+		}
+	}
+
 	if alias, ok := t.ackAliases[packet.Ack]; ok {
 		packet.UpdateAck(alias)
 	}
@@ -575,8 +723,8 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 
 		if !isIncoming {
 			if responseRequest != nil {
-				message.AssocMessage = responseRequest
-				responseRequest.AssocMessage = message
+				message.setAssocMessage(responseRequest)
+				responseRequest.setAssocMessage(message)
 			} else {
 				t.respWithoutReq[packet.Ack] = packet.ID
 			}
@@ -587,17 +735,18 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 	message.AddPacket(packet)
 
 	// Handling Expect: 100-continue requests
-	if message.Is100Continue() {
-		seq := packet.Seq + uint32(len(packet.Data))
+	if message.expectType == httpExpect100Continue && len(message.packets) == message.headerPacket+1 {
+		seq := packet.Seq + uint32(message.Size())
 		t.seqWithData[seq] = packet.Ack
 		message.DataSeq = seq
+		message.complete = false
 
 		// In case if sequence packet came first
 		for _, m := range t.messages {
 			if m.Seq == seq {
 				t.deleteMessage(m)
 				if m.AssocMessage != nil {
-					message.AssocMessage = m.AssocMessage
+					message.setAssocMessage(m.AssocMessage)
 				}
 				// log.Println("2: Adding ack alias:", m.Ack, packet.Ack)
 				t.ackAliases[m.Ack] = packet.Ack
@@ -626,15 +775,18 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 	}
 
 	// If message contains only single packet immediately dispatch it
-	if message.IsFinished() {
+	if message.complete {
+		// log.Println("COMPLETE!", isIncoming, message)
 		if isIncoming {
-			// log.Println("I'm finished", string(message.Bytes()), message.ResponseID, t.messages)
 			if t.trackResponse {
+				// log.Println("Found response!", message.ResponseID, t.messages)
+
 				if resp, ok := t.messages[message.ResponseID]; ok {
-					t.dispatchMessage(message)
-					if resp.IsFinished() {
+					if resp.complete {
 						t.dispatchMessage(resp)
 					}
+
+					t.dispatchMessage(message)
 				}
 			} else {
 				t.dispatchMessage(message)
@@ -645,7 +797,7 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
 			}
 
 			if req, ok := t.messages[message.AssocMessage.ID()]; ok {
-				if req.IsFinished() {
+				if req.complete {
 					t.dispatchMessage(req)
 					t.dispatchMessage(message)
 				}

raw_socket_listener/listener_test.go

@@ -46,6 +46,46 @@ func TestRawListenerInput(t *testing.T) {
 	}
 }
 
+func TestRawListenerInputResponseByClose(t *testing.T) {
+	var req, resp *TCPMessage
+
+	listener := NewListener("", "0", EnginePcap, true, 10*time.Millisecond)
+	defer listener.Close()
+
+	reqPacket := buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\n\r\n"))
+
+	respAck := reqPacket.Seq + uint32(len(reqPacket.Data))
+	respPacket := buildPacket(false, respAck, reqPacket.Seq+1, []byte("HTTP/1.1 200 OK\r\nConnection: close\r\n\r\nasd"))
+	finPacket := buildPacket(false, respAck, reqPacket.Seq+2, []byte(""))
+	finPacket.IsFIN = true
+
+	listener.packetsChan <- reqPacket.Dump()
+	listener.packetsChan <- respPacket.Dump()
+	listener.packetsChan <- finPacket.Dump()
+
+	select {
+	case req = <-listener.messagesChan:
+	case <-time.After(time.Millisecond):
+		t.Error("Should return request immediately")
+		return
+	}
+
+	if !req.IsIncoming {
+		t.Error("Should be request")
+	}
+
+	select {
+	case resp = <-listener.messagesChan:
+	case <-time.After(20 * time.Millisecond):
+		t.Error("Should return response immediately")
+		return
+	}
+
+	if resp.IsIncoming {
+		t.Error("Should be response")
+	}
+}
+
 func TestRawListenerInputWithoutResponse(t *testing.T) {
 	var req *TCPMessage
 
@@ -117,10 +157,10 @@ func TestShort100Continue(t *testing.T) {
 	reqPacket2 := buildPacket(true, 2, reqPacket1.Seq+uint32(len(reqPacket1.Data)), []byte("a"))
 	reqPacket3 := buildPacket(true, 2, reqPacket2.Seq+1, []byte("b"))
 
-	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n"))
+	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n\r\n"))
 
 	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
-	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
+	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n\r\n"))
 
 	result := []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\n\r\nab")
 
@@ -140,7 +180,7 @@ func Test100ContinueWrongOrder(t *testing.T) {
 	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n"))
 
 	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
-	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
+	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n\r\n"))
 
 	result := []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\n\r\nab")
 
@@ -159,15 +199,13 @@ func TestAlt100ContinueHeaderOrder(t *testing.T) {
 	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n"))
 
 	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
-	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
+	respPacket2 := buildPacket(false, reqPacket3.Seq+1 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n\r\n"))
 
 	result := []byte("POST / HTTP/1.1\r\nContent-Length: 2\r\n\r\nab")
 
 	testRawListener100Continue(t, listener, result, reqPacket1, reqPacket2, reqPacket3, respPacket1, respPacket2)
 }
 
-
-
 func testRawListener100Continue(t *testing.T, listener *Listener, result []byte, packets ...*TCPPacket) {
 	var req, resp *TCPMessage
 	for _, p := range packets {
@@ -320,17 +358,17 @@ func TestRawListenerChunkedWrongOrder(t *testing.T) {
 	reqPacket3 := buildPacket(true, 2, reqPacket2.Seq+uint32(len(reqPacket2.Data)), []byte("1\r\nb\r\n"))
 	reqPacket4 := buildPacket(true, 2, reqPacket3.Seq+uint32(len(reqPacket3.Data)), []byte("0\r\n\r\n"))
 
-	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n"))
+	respPacket1 := buildPacket(false, 10, 3, []byte("HTTP/1.1 100 Continue\r\n\r\n"))
 
 	// panic(int(uint32(len(reqPacket1.Data)) + uint32(len(reqPacket2.Data)) + uint32(len(reqPacket3.Data))))
-	respPacket2 := buildPacket(false, reqPacket4.Seq+5 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n"))
+	respPacket2 := buildPacket(false, reqPacket4.Seq+5 /* len of data */, 2, []byte("HTTP/1.1 200 OK\r\n\r\n"))
 
 	// Should re-construct message from all possible combinations
 	for i := 0; i < 6*5*4*3*2*1; i++ {
 
-		if i < 54 || i > 57 {
-			continue
-		}
+		// if i < 54 || i > 57 {
+		// 	continue
+		// }
 
 		packets := permutation(i, []*TCPPacket{reqPacket1, reqPacket2, reqPacket3, reqPacket4, respPacket1, respPacket2})
 
@@ -349,7 +387,7 @@ func chunkedPostMessage() []*TCPPacket {
 	reqPacket3 := buildPacket(true, ack, reqPacket2.Seq+5, []byte("1\r\nb\r\n"))
 	reqPacket4 := buildPacket(true, ack, reqPacket3.Seq+5, []byte("0\r\n\r\n"))
 
-	respPacket := buildPacket(false, reqPacket4.Seq+5 /* len of data */, ack, []byte("HTTP/1.1 200 OK\r\n"))
+	respPacket := buildPacket(false, reqPacket4.Seq+5 /* len of data */, ack, []byte("HTTP/1.1 200 OK\r\n\r\n"))
 
 	return []*TCPPacket{
 		reqPacket1, reqPacket2, reqPacket3, reqPacket4, respPacket,
@@ -372,7 +410,7 @@ func postMessage() []*TCPPacket {
 
 	return []*TCPPacket{
 		buildPacket(true, ack, seq, data),
-		buildPacket(false, seq+uint32(len(data)), seq2, []byte("HTTP/1.1 200 OK\r\n")),
+		buildPacket(false, seq+uint32(len(data)), seq2, []byte("HTTP/1.1 200 OK\r\n\r\n")),
 	}
 }
 
@@ -383,7 +421,7 @@ func getMessage() []*TCPPacket {
 
 	return []*TCPPacket{
 		buildPacket(true, ack, seq, []byte("GET / HTTP/1.1\r\n\r\n")),
-		buildPacket(false, seq+18, seq2, []byte("HTTP/1.1 200 OK\r\n")),
+		buildPacket(false, seq+18, seq2, []byte("HTTP/1.1 200 OK\r\n\r\n")),
 	}
 }
 

raw_socket_listener/tcp_message.go

@@ -7,9 +7,9 @@ import (
 	"encoding/hex"
 	"github.com/buger/gor/proto"
 	"log"
+	"net"
 	"strconv"
 	"time"
-	"net"
 )
 
 var _ = log.Println
@@ -36,6 +36,15 @@ type TCPMessage struct {
 	packets []*TCPPacket
 
 	delChan chan *TCPMessage
+
+	/* HTTP specific variables */
+	methodType    httpMethodType
+	bodyType      httpBodyType
+	expectType    httpExpectType
+	seqMissing    bool
+	headerPacket  int
+	contentLength int
+	complete      bool
 }
 
 // NewTCPMessage pointer created from a Acknowledgment number and a channel of messages readuy to be deleted
@@ -46,6 +55,15 @@ func NewTCPMessage(Seq, Ack uint32, IsIncoming bool) (msg *TCPMessage) {
 	return
 }
 
+func (t *TCPMessage) packetsData() (d [][]byte) {
+	d = make([][]byte, len(t.packets))
+	for i, p := range t.packets {
+		d[i] = p.Data
+	}
+
+	return
+}
+
 // Bytes return message content
 func (t *TCPMessage) Bytes() (output []byte) {
 	for _, p := range t.packets {
@@ -57,13 +75,13 @@ func (t *TCPMessage) Bytes() (output []byte) {
 
 // Size returns total body size
 func (t *TCPMessage) BodySize() (size int) {
-	if len(t.packets) == 0 {
+	if len(t.packets) == 0 || t.headerPacket == -1 {
 		return 0
 	}
 
-	size += len(proto.Body(t.packets[0].Data))
+	size += len(proto.Body(t.packets[t.headerPacket].Data))
 
-	for _, p := range t.packets[1:] {
+	for _, p := range t.packets[t.headerPacket+1:] {
 		size += len(p.Data)
 	}
 
@@ -121,139 +139,344 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) {
 			t.DataAck = packet.OrigAck
 		}
 	}
+
+	t.checkSeqIntegrity()
+	t.updateHeadersPacket()
+	t.updateMethodType()
+	t.updateBodyType()
+	t.checkIfComplete()
+	t.check100Continue()
 }
 
 // Check if there is missing packet
-func (t *TCPMessage) isSeqMissing() bool {
+func (t *TCPMessage) checkSeqIntegrity() {
 	if len(t.packets) == 1 {
-		return false
+		t.seqMissing = false
 	}
 
-	for i, p := range t.packets {
+	offset := len(t.packets) - 1
+
+	if t.packets[offset].IsFIN {
+		offset--
+
+		if offset < 0 {
+			return
+		}
+	}
+
+	for i, p := range t.packets[:offset] {
+		if p.IsFIN {
+			continue
+		}
+
 		// If final packet
 		if len(t.packets) == i+1 {
-			return false
+			t.seqMissing = false
+			return
 		}
 		np := t.packets[i+1]
 
-		if np.Seq != p.Seq+uint32(len(p.Data)) {
-			return true
+		nextSeq := p.Seq + uint32(len(p.Data))
+
+		if np.Seq != nextSeq {
+			if t.expectType == httpExpect100Continue {
+				if np.Seq != nextSeq+22 {
+					t.seqMissing = true
+					return
+				}
+			} else {
+				t.seqMissing = true
+				return
+			}
 		}
 	}
 
-	return false
+	t.seqMissing = false
 }
 
-var EmptyLine = []byte("\r\n\r\n")
-var ChunkEnd = []byte("0\r\n\r\n")
+var bEmptyLine = []byte("\r\n\r\n")
+var bChunkEnd = []byte("0\r\n\r\n")
 
-func (t *TCPMessage) isHeadersReceived() bool {
-	for _, p := range t.packets {
-		if bytes.LastIndex(p.Data, EmptyLine) != -1 {
-			return true
+func (t *TCPMessage) updateHeadersPacket() {
+	if len(t.packets) == 1 {
+		t.headerPacket = -1
+	}
+
+	if t.headerPacket != -1 {
+		return
+	}
+
+	if t.seqMissing {
+		return
+	}
+
+	for i, p := range t.packets {
+		if bytes.LastIndex(p.Data, bEmptyLine) != -1 {
+			t.headerPacket = i
+			return
 		}
 	}
 
-	return false
+	return
 }
 
 // isMultipart returns true if message contains from multiple tcp packets
-func (t *TCPMessage) IsFinished() bool {
-	payload := t.packets[0].Data
-
-	if len(payload) < 4 {
-		return true
+func (t *TCPMessage) checkIfComplete() {
+	if t.seqMissing || t.headerPacket == -1 {
+		return
 	}
 
-	m := payload[:4]
+	if t.methodType == httpMethodNotFound {
+		return
+	}
 
-	if t.IsIncoming {
-		// If one GET, OPTIONS, or HEAD request
-		if bytes.Equal(m, []byte("GET ")) || bytes.Equal(m, []byte("OPTI")) || bytes.Equal(m, []byte("HEAD")) {
-			if !t.isSeqMissing() && t.isHeadersReceived() {
-				return true
-			} else {
-				return false
-			}
-		} else {
-			// Sometimes header comes after the body :(
-			if bytes.Equal(m, []byte("POST")) || bytes.Equal(m, []byte("PUT ")) || bytes.Equal(m, []byte("PATC")) {
+	// Responses can be emitted only if we found request
+	if !t.IsIncoming && t.AssocMessage == nil {
+		return
+	}
 
-				if t.isHeadersReceived() {
-					if length := proto.Header(payload, []byte("Content-Length")); len(length) > 0 {
-						l, _ := strconv.Atoi(string(length))
-
-						// If content-length equal current body length
-						if l > 0 && l == t.BodySize() {
-							return true
-						}
-					}
-				}
-			}
-		}
+	// If one GET, OPTIONS, or HEAD request
+	if t.methodType == httpMethodWithoutBody {
+		t.complete = true
 	} else {
-		// Request not found
-		// Can be because response came first or request request was just missing
-		if t.AssocMessage == nil {
-			return false
-		}
-
-		if !bytes.Equal(m, []byte("HTTP")) {
-			return false
-		}
-
-		if length := proto.Header(payload, []byte("Content-Length")); len(length) > 0 {
-			if length[0] == '0' {
-				return true
+		switch t.bodyType {
+		case httpBodyEmpty:
+			t.complete = true
+		case httpBodyContentLength:
+			if t.contentLength == 0 || t.contentLength == t.BodySize() {
+				t.complete = true
+			}
+		case httpBodyChunked:
+			lastPacket := t.packets[len(t.packets)-1]
+			if bytes.LastIndex(lastPacket.Data, bChunkEnd) != -1 {
+				t.complete = true
+			}
+		default:
+			if len(t.packets) == 0 {
+				return
 			}
 
-			l, _ := strconv.Atoi(string(length))
-
-			// If content-length equal current body length
-			if l > 0 && l == t.BodySize() {
-				return true
-			}
-		} else {
-			if enc := proto.Header(payload, []byte("Transfer-Encoding")); len(enc) == 0 {
-				return true
-			} else {
-				if len(t.packets) > 1 && bytes.LastIndex(t.packets[len(t.packets)-1].Data, ChunkEnd) != -1 {
-					return true
-				}
+			last := t.packets[len(t.packets)-1]
+			if last.IsFIN {
+				t.complete = true
 			}
 		}
 	}
-
-	return false
 }
 
-var bExpectHeader = []byte("Expect:")
-var bExpect100Value = []byte("100-continue")
-var bPOST = []byte("POST")
-var bCRLFx2 = []byte("\r\n\r\n")
+type httpMethodType uint8
+
+const (
+	httpMethodNotSet      httpMethodType = 0
+	httpMethodWithBody    httpMethodType = 1
+	httpMethodWithoutBody httpMethodType = 2
+	httpMethodNotFound    httpMethodType = 3
+)
+
+var methodsWithBody = [][]byte{
+	[]byte("POST"),
+	[]byte("PUT"),
+	[]byte("PATCH"),
+	[]byte("CONNECT"),
+}
+
+func (t *TCPMessage) updateMethodType() {
+	// if there is cache
+	if t.methodType != httpMethodNotSet && t.methodType != httpMethodNotFound {
+		return
+	}
 
-func (t *TCPMessage) Is100Continue() bool {
 	d := t.packets[0].Data
 
-	if len(d) < 25 {
-		return false
+	// Minimum length fo request: GET / HTTP/1.1\r\n
+
+	if len(d) < 16 {
+		t.methodType = httpMethodNotFound
+		return
 	}
 
-	if !bytes.Equal(d[0:4], bPOST) {
-		return false
+	if t.IsIncoming {
+		var method []byte
+
+		if mIdx := bytes.IndexByte(d[:8], ' '); mIdx != -1 {
+			method = d[:mIdx]
+
+			// Check that after method we have absolute or relative path
+			switch d[mIdx+1] {
+			case '/', 'h', '*':
+			default:
+				t.methodType = httpMethodNotFound
+				return
+			}
+		} else {
+			t.methodType = httpMethodNotFound
+			return
+		}
+
+		for _, m := range methodsWithBody {
+			if len(m) == len(method) && bytes.Equal(m, method) {
+				t.methodType = httpMethodWithBody
+				return
+			}
+		}
+
+		t.methodType = httpMethodWithoutBody
+	} else {
+		if !bytes.Equal(d[:6], []byte("HTTP/1")) {
+			t.methodType = httpMethodNotFound
+			return
+		}
+
+		t.methodType = httpMethodWithBody
+	}
+}
+
+type httpBodyType uint8
+
+const (
+	httpBodyNotSet          httpBodyType = 0
+	httpBodyEmpty           httpBodyType = 1
+	httpBodyContentLength   httpBodyType = 2
+	httpBodyChunked         httpBodyType = 3
+	httpBodyConnectionClose httpBodyType = 4
+)
+
+func (t *TCPMessage) updateBodyType() {
+	// if there is cache
+	if t.bodyType != httpBodyNotSet {
+		return
 	}
 
+	// Headers not received
+	if t.headerPacket == -1 {
+		return
+	}
+
+	var lengthB, encB, connB []byte
+
+	proto.ParseHeaders(t.packetsData(), func(header, value []byte)bool{
+		if proto.HeadersEqual(header, []byte("Content-Length")) {
+			lengthB = value
+			return false
+		}
+
+		if proto.HeadersEqual(header, []byte("Transfer-Encoding")) {
+			encB = value
+			return false
+		}
+
+		if proto.HeadersEqual(header, []byte("Connection")) {
+			connB = value
+		}
+
+		return true
+	})
+
+	switch t.methodType {
+	case httpMethodNotFound:
+		return
+	case httpMethodWithoutBody:
+		t.bodyType = httpBodyEmpty
+		return
+	case httpMethodWithBody:
+		if len(lengthB) > 0 {
+			t.contentLength, _ = strconv.Atoi(string(lengthB))
+
+			if t.contentLength == 0 {
+				t.bodyType = httpBodyEmpty
+			} else {
+				t.bodyType = httpBodyContentLength
+			}
+			return
+		}
+
+		if len(encB) > 0 {
+			t.bodyType = httpBodyChunked
+			return
+		}
+
+		if len(connB) > 0 && bytes.Equal(connB, []byte("close")) {
+			t.bodyType = httpBodyConnectionClose
+			return
+		}
+	}
+
+	t.bodyType = httpBodyEmpty
+}
+
+type httpExpectType uint8
+
+const (
+	httpExpectNotSet      httpExpectType = 0
+	httpExpectEmpty       httpExpectType = 1
+	httpExpect100Continue httpExpectType = 2
+)
+
+var bExpectHeader = []byte("Expect")
+var bExpect100Value = []byte("100-continue")
+
+func (t *TCPMessage) check100Continue() {
+	if t.expectType != httpExpectNotSet || len(t.packets[0].Data) < 25 {
+		return
+	}
+
+	if t.methodType != httpMethodWithBody {
+		return
+	}
+
+	if t.seqMissing || t.headerPacket == -1 {
+		return
+	}
+
+	last := t.packets[len(t.packets)-1]
 	// reading last 4 bytes for double CRLF
-	if !bytes.Equal(d[len(d)-4:], bCRLFx2) {
-		return false
+	if !bytes.HasSuffix(last.Data, bEmptyLine) {
+		return
 	}
 
-	// look for an expect:100-continue header
-	if !bytes.Equal(bExpect100Value, proto.Header(d, bExpectHeader)) {
-		return false
+	var expectB []byte
+	proto.ParseHeaders(t.packetsData(), func(header, value []byte)bool{
+		if proto.HeadersEqual(header, bExpectHeader) {
+			expectB = value
+			return false
+		}
+
+		return true
+	})
+
+	if len(expectB) > 0 && bytes.Equal(bExpect100Value, expectB) {
+		t.expectType = httpExpect100Continue
+		return
 	}
 
-	return true
+	t.expectType = httpExpectEmpty
+}
+
+func (t *TCPMessage) setAssocMessage(m *TCPMessage) {
+	t.AssocMessage = m
+	t.checkIfComplete()
+}
+
+// UpdateResponseAck should be called after packet is added
+func (t *TCPMessage) UpdateResponseAck() uint32 {
+	lastPacket := t.packets[len(t.packets)-1]
+	if lastPacket.IsFIN && len(t.packets) > 1 {
+		lastPacket = t.packets[len(t.packets)-2]
+	}
+
+	respAck := lastPacket.Seq + uint32(len(lastPacket.Data))
+
+	if t.ResponseAck != respAck {
+		t.ResponseAck = lastPacket.Seq + uint32(len(lastPacket.Data))
+
+		// We swappwed src and dst port
+		copy(t.ResponseID[:16], lastPacket.Addr)
+		copy(t.ResponseID[16:], lastPacket.Raw[2:4]) // Src port
+		copy(t.ResponseID[18:], lastPacket.Raw[0:2]) // Dest port
+		binary.BigEndian.PutUint32(t.ResponseID[20:24], t.ResponseAck)
+	}
+
+	return t.ResponseAck
 }
 
 func (t *TCPMessage) UUID() []byte {
@@ -276,28 +499,10 @@ func (t *TCPMessage) UUID() []byte {
 	return uuid
 }
 
-// UpdateResponseAck should be called after packet is added
-func (t *TCPMessage) UpdateResponseAck() uint32 {
-	lastPacket := t.packets[len(t.packets)-1]
-	respAck := lastPacket.Seq + uint32(len(lastPacket.Data))
-
-	if t.ResponseAck != respAck {
-		t.ResponseAck = lastPacket.Seq + uint32(len(lastPacket.Data))
-
-		// We swappwed src and dst port
-		copy(t.ResponseID[:16], lastPacket.Addr)
-		copy(t.ResponseID[16:], lastPacket.Raw[2:4]) // Src port
-		copy(t.ResponseID[18:], lastPacket.Raw[0:2]) // Dest port
-		binary.BigEndian.PutUint32(t.ResponseID[20:24], t.ResponseAck)
-	}
-
-	return t.ResponseAck
-}
-
 func (t *TCPMessage) ID() tcpID {
 	return t.packets[0].ID
 }
 
 func (t *TCPMessage) IP() net.IP {
 	return net.IP(t.packets[0].Addr)
-}
+}

raw_socket_listener/tcp_message_test.go

@@ -80,77 +80,45 @@ func TestTCPMessageSize(t *testing.T) {
 	}
 }
 
-func TestTCPMessageIsFinished(t *testing.T) {
-	methodsWithoutBodies := []string{"GET", "OPTIONS", "HEAD"}
+func TestTCPMessageIsComplete(t *testing.T) {
+	testCases := []struct {
+		direction         bool
+		payload           string
+		assocMessage      bool
+		expectedCompleted bool
+	}{
+		{true, "GET / HTTP/1.1\r\n\r\n", false, true},
+		{true, "HEAD / HTTP/1.1\r\n\r\n", false, true},
+		{false, "HTTP/1.1 200 OK\r\n\r\n", true, true},
+		{true, "POST / HTTP/1.1\r\nContent-Length: 1\r\n\r\na", false, true},
+		{true, "PUT / HTTP/1.1\r\nContent-Length: 1\r\n\r\na", false, true},
+		{false, "HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n", true, true},
+		{false, "HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\na", true, true},
+		{false, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n0\r\n\r\n", true, true},
 
-	for _, m := range methodsWithoutBodies {
-		msg := buildMessage(buildPacket(true, 1, 1, []byte(m+" / HTTP/1.1\r\n\r\n")))
+		// chunked not finished
+		{false, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n", true, false},
 
-		if !msg.IsFinished() {
-			t.Error(m, " request should be finished")
+		// content-length != actual length
+		{true, "POST / HTTP/1.1\r\nContent-Length: 2\r\n\r\na", false, false},
+		{false, "HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\na", true, false},
+		// non-valid http request
+		{true, "UNKNOWN asd HTTP/1.1\r\n\r\n", false, false},
+
+		// response without associated request
+		{false, "HTTP/1.1 200 OK\r\n\r\n", false, false},
+	}
+
+	for _, tc := range testCases {
+		msg := buildMessage(buildPacket(tc.direction, 1, 1, []byte(tc.payload)))
+		if tc.assocMessage {
+			msg.AssocMessage = &TCPMessage{}
 		}
-	}
+		msg.checkIfComplete()
 
-	methodsWithBodies := []string{"POST", "PUT", "PATCH"}
-
-	for _, m := range methodsWithBodies {
-		msg := buildMessage(buildPacket(true, 1, 1, []byte(m+" / HTTP/1.1\r\nContent-Length: 1\r\n\r\na")))
-
-		if !msg.IsFinished() {
-			t.Error(m, " should be finished as body length == content length")
+		if msg.complete != tc.expectedCompleted {
+			t.Errorf("Payload %s: Expected %t, got %t.", tc.payload, tc.expectedCompleted, msg.complete)
 		}
-
-		msg = buildMessage(buildPacket(true, 1, 1, []byte(m+" / HTTP/1.1\r\nContent-Length: 2\r\n\r\na")))
-
-		if msg.IsFinished() {
-			t.Error(m, " should not be finished as body length != content length")
-		}
-	}
-
-	msg := buildMessage(buildPacket(true, 1, 1, []byte("UNKNOWN / HTTP/1.1\r\n\r\n")))
-	if msg.IsFinished() {
-		t.Error("non http or wrong methods considered as not finished")
-	}
-
-	// Responses
-	msg = buildMessage(buildPacket(false, 1, 1, []byte("HTTP/1.1 200 OK\r\n\r\n")))
-	msg.AssocMessage = &TCPMessage{}
-	if !msg.IsFinished() {
-		t.Error("Should mark simple response as finished")
-	}
-
-	msg = buildMessage(buildPacket(false, 1, 1, []byte("HTTP/1.1 200 OK\r\n\r\n")))
-	msg.AssocMessage = nil
-	if msg.IsFinished() {
-		t.Error("Should not mark responses without associated requests")
-	}
-
-	msg = buildMessage(buildPacket(false, 1, 1, []byte("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n")))
-	msg.AssocMessage = &TCPMessage{}
-
-	if msg.IsFinished() {
-		t.Error("Should mark chunked response as non finished")
-	}
-
-	msg = buildMessage(buildPacket(false, 1, 1, []byte("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")))
-	msg.AssocMessage = &TCPMessage{}
-
-	if !msg.IsFinished() {
-		t.Error("Should mark Content-Length: 0 respones as finished")
-	}
-
-	msg = buildMessage(buildPacket(false, 1, 1, []byte("HTTP/1.1 200 OK\r\nContent-Length: 1\r\n\r\na")))
-	msg.AssocMessage = &TCPMessage{}
-
-	if !msg.IsFinished() {
-		t.Error("Should mark valid Content-Length respones as finished")
-	}
-
-	msg = buildMessage(buildPacket(false, 1, 1, []byte("HTTP/1.1 200 OK\r\nContent-Length: 10\r\n\r\na")))
-	msg.AssocMessage = &TCPMessage{}
-
-	if msg.IsFinished() {
-		t.Error("Should not mark not valid Content-Length respones as finished")
 	}
 }
 
@@ -160,32 +128,118 @@ func TestTCPMessageIsSeqMissing(t *testing.T) {
 	p3 := buildPacket(false, 1, p2.Seq+uint32(len(p2.Data)), []byte("a"))
 
 	msg := buildMessage(p1)
-	if msg.isSeqMissing() {
+	if msg.seqMissing {
 		t.Error("Should be complete if have only 1 packet")
 	}
 
 	msg.AddPacket(p3)
-	if !msg.isSeqMissing() {
+	if !msg.seqMissing {
 		t.Error("Should be incomplete because missing middle component")
 	}
 
 	msg.AddPacket(p2)
-	if msg.isSeqMissing() {
+	if msg.seqMissing {
 		t.Error("Should be complete once missing packet added")
 	}
 }
 
 func TestTCPMessageIsHeadersReceived(t *testing.T) {
-	p1 := buildPacket(false, 1, 1, []byte("HTTP/1.1 200 OK\r\n"))
+	p1 := buildPacket(false, 1, 1, []byte("HTTP/1.1 200 OK\r\n\r\n"))
 	p2 := buildPacket(false, 1, p1.Seq+uint32(len(p1.Data)), []byte("Content-Length: 10\r\n\r\n"))
 
 	msg := buildMessage(p1)
-	if msg.isHeadersReceived() {
-		t.Error("Should be complete if have only 1 packet")
+	if msg.headerPacket == -1 {
+		t.Error("Should be complete if have only 1 packet", msg.headerPacket)
 	}
 
 	msg.AddPacket(p2)
-	if !msg.isHeadersReceived() {
+	if msg.headerPacket == -1 {
 		t.Error("Should found double new line: headers received")
 	}
+
+	msg = buildMessage(buildPacket(true, 1, 1, []byte("GET / HTTP/1.1\r\nContent-Length: 1\r\n")))
+	if msg.headerPacket != -1 {
+		t.Error("Should not find headers end")
+	}
+}
+
+func TestTCPMessageMethodType(t *testing.T) {
+	testCases := []struct {
+		direction          bool
+		payload            string
+		expectedMethodType httpMethodType
+	}{
+		{true, "GET / HTTP/1.1\r\n\r\n", httpMethodWithoutBody},
+		{true, "GET * HTTP/1.1\r\n\r\n", httpMethodWithoutBody},
+		{true, "UNKNOWN / HTTP/1.1\r\n\r\n", httpMethodWithoutBody},
+		{true, "GET http://example.com HTTP/1.1\r\n\r\n", httpMethodWithoutBody},
+		{true, "POST / HTTP/1.1\r\n\r\n", httpMethodWithBody},
+		{true, "PUT / HTTP/1.1\r\n\r\n", httpMethodWithBody},
+		{true, "GET zxc HTTP/1.1\r\n\r\n", httpMethodNotFound},
+		{true, "GET / HTTP\r\n\r\n", httpMethodNotFound},
+		{true, "VERYLONGMETHOD / HTTP/1.1\r\n\r\n", httpMethodNotFound},
+		{false, "HTTP/1.1 200 OK\r\n\r\n", httpMethodWithBody},
+		{false, "HTTP /1.1 200 OK\r\n\r\n", httpMethodNotFound},
+	}
+
+	for _, tc := range testCases {
+		msg := buildMessage(buildPacket(tc.direction, 1, 1, []byte(tc.payload)))
+
+		if msg.methodType != tc.expectedMethodType {
+			t.Errorf("Expected %d, got %d", tc.expectedMethodType, msg.methodType)
+		}
+	}
+}
+
+func TestTCPMessageBodyType(t *testing.T) {
+	testCases := []struct {
+		direction        bool
+		payload          string
+		expectedBodyType httpBodyType
+	}{
+		{true, "GET / HTTP/1.1\r\n\r\n", httpBodyEmpty},
+		{true, "POST / HTTP/1.1\r\n\r\n", httpBodyEmpty},
+		{true, "POST / HTTP/1.1\r\nUser-Agent: zxc\r\n\r\n", httpBodyEmpty},
+		{false, "HTTP/1.1 200 OK\r\n\r\n", httpBodyEmpty},
+		{true, "POST / HTTP/1.1\r\nContent-Length: 2\r\n\r\nab", httpBodyContentLength},
+		{false, "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nab", httpBodyContentLength},
+		{true, "POST / HTTP/1.1\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nab\r\n0\r\n\r\n", httpBodyChunked},
+		{false, "HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n2\r\nab\r\n0\r\n\r\n", httpBodyChunked},
+	}
+
+	for _, tc := range testCases {
+		msg := buildMessage(buildPacket(tc.direction, 1, 1, []byte(tc.payload)))
+
+		if msg.bodyType != tc.expectedBodyType {
+			t.Errorf("Expected %d, got %d", tc.expectedBodyType, msg.bodyType)
+		}
+	}
+}
+
+func TestTCPMessageBodySize(t *testing.T) {
+	testCases := []struct {
+		direction    bool
+		payloads     []string
+		expectedSize int
+	}{
+		{true, []string{"GET / HTTP/1.1\r\n\r\n"}, 0},
+		{true, []string{"POST / HTTP/1.1\r\nContent-Length: 2\r\n\r\nab"}, 2},
+		{true, []string{"GET / HTTP/1.1\r\n", "Content-Length: 2\r\n\r\nab"}, 2},
+		{true, []string{"GET / HTTP/1.1\r\n", "Content-Length: 2\r\n\r\n", "ab"}, 2},
+	}
+
+	for _, tc := range testCases {
+		msg := buildMessage(buildPacket(tc.direction, 1, 1, []byte(tc.payloads[0])))
+
+		if len(tc.payloads) > 1 {
+			for _, p := range tc.payloads[1:] {
+				seq := uint32(1 + msg.Size())
+				msg.AddPacket(buildPacket(tc.direction, 1, seq, []byte(p)))
+			}
+		}
+
+		if msg.BodySize() != tc.expectedSize {
+			t.Errorf("Expected %d, got %d", tc.expectedSize, msg.BodySize())
+		}
+	}
 }

raw_socket_listener/tcp_packet.go

@@ -33,6 +33,7 @@ type TCPPacket struct {
 	Ack        uint32
 	OrigAck    uint32
 	DataOffset uint8
+	IsFIN      bool
 
 	Raw  []byte
 	Data []byte
@@ -71,6 +72,7 @@ func (t *TCPPacket) ParseBasic() {
 	t.Seq = binary.BigEndian.Uint32(t.Raw[4:8])
 	t.Ack = binary.BigEndian.Uint32(t.Raw[8:12])
 	t.DataOffset = (t.Raw[12] & 0xF0) >> 4
+	t.IsFIN = t.Raw[13]&0x01 != 0
 
 	// log.Println("DataOffset:", t.DataOffset, t.DestPort, t.SrcPort, t.Seq, t.Ack)
 
@@ -90,6 +92,11 @@ func (t *TCPPacket) Dump() []byte {
 	binary.BigEndian.PutUint32(tcpBuf[8:12], t.Ack)
 
 	tcpBuf[12] = 64
+
+	if t.IsFIN {
+		tcpBuf[13] = tcpBuf[13] | 0x01
+	}
+
 	copy(tcpBuf[16:], t.Data)
 
 	return buf
@@ -109,6 +116,7 @@ func (t *TCPPacket) String() string {
 		"Sequence:" + strconv.Itoa(int(t.Seq)),
 		"Acknowledgment:" + strconv.Itoa(int(t.Ack)),
 		"Header len:" + strconv.Itoa(int(t.DataOffset)),
+		"FIN:" + strconv.FormatBool(t.IsFIN),
 
 		"Data size:" + strconv.Itoa(len(t.Data)),
 		"Data:" + string(t.Data[:maxLen]),

settings.go

@@ -25,15 +25,17 @@ func (h *MultiOption) Set(value string) error {
 
 // AppSettings is the struct of main configuration
 type AppSettings struct {
-	verbose bool
-	debug   bool
-	stats   bool
+	verbose   bool
+	debug     bool
+	stats     bool
+	exitAfter time.Duration
 
 	splitOutput bool
 
 	inputDummy   MultiOption
 	outputDummy  MultiOption
 	outputStdout bool
+	outputNull   bool
 
 	inputTCP       MultiOption
 	outputTCP      MultiOption
@@ -73,6 +75,7 @@ func init() {
 	flag.BoolVar(&Settings.verbose, "verbose", false, "Turn on more verbose output")
 	flag.BoolVar(&Settings.debug, "debug", false, "Turn on debug output, shows all intercepted traffic. Works only when with `verbose` flag")
 	flag.BoolVar(&Settings.stats, "stats", false, "Turn on queue stats output")
+	flag.DurationVar(&Settings.exitAfter, "exit-after", 0, "exit after specified duration")
 
 	flag.BoolVar(&Settings.splitOutput, "split-output", false, "By default each output gets same traffic. If set to `true` it splits traffic equally among all outputs.")
 
@@ -81,6 +84,8 @@ func init() {
 
 	flag.BoolVar(&Settings.outputStdout, "output-stdout", false, "Used for testing inputs. Just prints to console data coming from inputs.")
 
+	flag.BoolVar(&Settings.outputNull, "output-null", false, "Used for testing inputs. Drops all requests.")
+
 	flag.Var(&Settings.inputTCP, "input-tcp", "Used for internal communication between Gor instances. Example: \n\t# Receive requests from other Gor instances on 28020 port, and redirect output to staging\n\tgor --input-tcp :28020 --output-http staging.com")
 	flag.Var(&Settings.outputTCP, "output-tcp", "Used for internal communication between Gor instances. Example: \n\t# Listen for requests on 80 port and forward them to other Gor instance on 28020 port\n\tgor --input-raw :80 --output-tcp replay.local:28020")
 	flag.BoolVar(&Settings.outputTCPStats, "output-tcp-stats", false, "Report TCP output queue stats to console every 5 seconds.")
@@ -107,7 +112,7 @@ func init() {
 
 	flag.StringVar(&Settings.middleware, "middleware", "", "Used for modifying traffic using external command")
 
-	flag.Var(&Settings.inputHTTP, "input-http", "Read requests from HTTP, should be explicitly sent from your application:\n\t# Listen for http on 9000\n\tgor --input-http :9000 --output-http staging.com")
+	// flag.Var(&Settings.inputHTTP, "input-http", "Read requests from HTTP, should be explicitly sent from your application:\n\t# Listen for http on 9000\n\tgor --input-http :9000 --output-http staging.com")
 
 	flag.Var(&Settings.outputHTTP, "output-http", "Forwards incoming requests to given http address.\n\t# Redirect all incoming requests to staging.com address \n\tgor --input-raw :80 --output-http http://staging.com")
 	flag.IntVar(&Settings.outputHTTPConfig.BufferSize, "output-http-response-buffer", 0, "HTTP response buffer size, all data after this size will be discarded.")
@@ -140,7 +145,7 @@ func init() {
 	flag.Var(&Settings.modifierConfig.headerFilters, "http-allow-header", "A regexp to match a specific header against. Requests with non-matching headers will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-allow-header api-version:^v1")
 	flag.Var(&Settings.modifierConfig.headerFilters, "output-http-header-filter", "WARNING: `--output-http-header-filter` DEPRECATED, use `--http-allow-header` instead")
 
-	flag.Var(&Settings.modifierConfig.headerFilters, "http-disallow-header", "A regexp to match a specific header against. Requests with matching headers will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-disallow-header \"User-Agent: Replayed by Gor\"")
+	flag.Var(&Settings.modifierConfig.headerNegativeFilters, "http-disallow-header", "A regexp to match a specific header against. Requests with matching headers will be dropped:\n\t gor --input-raw :8080 --output-http staging.com --http-disallow-header \"User-Agent: Replayed by Gor\"")
 
 	flag.Var(&Settings.modifierConfig.headerHashFilters, "http-header-limiter", "Takes a fraction of requests, consistently taking or rejecting a request based on the FNV32-1A hash of a specific header:\n\t gor --input-raw :8080 --output-http staging.com --http-header-imiter user-id:25%")
 	flag.Var(&Settings.modifierConfig.headerHashFilters, "output-http-header-hash-filter", "WARNING: `output-http-header-hash-filter` DEPRECATED, use `--http-header-hash-limiter` instead")

test_input.go

@@ -22,7 +22,7 @@ func NewTestInput() (i *TestInput) {
 func (i *TestInput) Read(data []byte) (int, error) {
 	buf := <-i.data
 
-	header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano())
+	header := payloadHeader(RequestPayload, uuid(), time.Now().UnixNano(), -1)
 	copy(data[0:len(header)], header)
 	copy(data[len(header):], buf)