update thrift, opencensus, others (#893)

* update thrift, opencensus, others

* stats: update to opencensus 0.6.0 view api
This commit is contained in:
Reed Allman
2018-03-26 15:43:49 -07:00
committed by GitHub
parent 0ce5c4500b
commit 8af605cf3d
1266 changed files with 122191 additions and 28775 deletions

View File

@@ -97,6 +97,10 @@ if WITH_RS
SUBDIRS += rs
endif
if WITH_CL
SUBDIRS += cl
endif
# All of the libs that don't use Automake need to go in here
# so they will end up in our release tarballs.
EXTRA_DIST = \

View File

@@ -0,0 +1,5 @@
# Maven Ant tasks Jar details
mvn.ant.task.version=2.1.3
mvn.repo=http://repo1.maven.org/maven2
mvn.ant.task.url=${mvn.repo}/org/apache/maven/maven-ant-tasks/${mvn.ant.task.version}
mvn.ant.task.jar=maven-ant-tasks-${mvn.ant.task.version}.jar

View File

@@ -7,9 +7,9 @@
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -19,23 +19,24 @@
-->
<project name="libthrift-as3" default="compile" basedir="."
xmlns:artifact="antlib:org.apache.maven.artifact.ant">
<property name="as3.artifactid" value="${ant.project.name}"/>
<property name="thrift.root" location="${basedir}/../../"/>
<property name="thrift.java.dir" location="${thrift.root}/lib/java"/>
<property name="build.tools.dir" location="${thrift.java.dir}/build/tools/"/>
<property name="thrift_compiler" value="${thrift.root}/compiler/cpp/thrift"/>
<property name="thrift_compiler" value="${thrift.root}/compiler/cpp/thrift"/>
<property file="${basedir}/build.properties"/>
<!-- inherit from the java build file for version and other properties -->
<property file="${thrift.java.dir}/build.properties" />
<property file="${thrift.java.dir}/gradle.properties" />
<property environment="env"/>
<condition property="version" value="${thrift.version}">
<isset property="release"/>
</condition>
<property name="version" value="${thrift.version}-snapshot"/>
<property name="version" value="${thrift.version}-SNAPSHOT"/>
<property name="as3.final.name" value="${as3.artifactid}-${version}"/>
@@ -52,12 +53,13 @@
<target name="setup.init">
<tstamp/>
<mkdir dir="${build.dir}"/>
<mkdir dir="${build.tools.dir}"/>
</target>
<target name="flex.check" unless="FLEX_HOME">
<fail message='You must set the FLEX_HOME property pointing to your flex SDK, eg. ant -DFLEX_HOME="/Applications/Adobe Flex Builder 3/sdks/3.2.0"'/>
</target>
<target name="flex.init" depends="flex.check" unless="flex.finished">
<taskdef resource="flexTasks.tasks" classpath="${FLEX_HOME}/ant/lib/flexTasks.jar" />
<property name="flex.finished" value="true"/>
@@ -69,10 +71,10 @@
<include name="**/*.as"/>
</fileset>
</path>
<pathconvert
property="as.src.classes"
pathsep=" "
dirsep="."
<pathconvert
property="as.src.classes"
pathsep=" "
dirsep="."
refid="as.src.files"
>
<map from="${src}/" to=""/>
@@ -108,19 +110,19 @@
<artifact:remoteRepository id="apache" url="${apache.repo}"/>
<!-- Pom file information -->
<artifact:pom id="pom"
groupId="${thrift.groupid}"
<artifact:pom id="pom"
groupId="${thrift.groupid}"
artifactId="${as3.artifactid}"
version="${version}"
version="${version}"
url="http://thrift.apache.org"
name="Apache Thrift"
description="Thrift is a software framework for scalable cross-language services development."
packaging="pom"
packaging="swc"
>
<remoteRepository refid="central"/>
<remoteRepository refid="apache"/>
<license name="The Apache Software License, Version 2.0" url="${license}"/>
<scm connection="scm:git:https://git-wip-us.apache.org/repos/asf/thrift.git"
<scm connection="scm:git:https://git-wip-us.apache.org/repos/asf/thrift.git"
developerConnection="scm:git:https://git-wip-us.apache.org/repos/asf/thrift.git"
url="https://git-wip-us.apache.org/repos/asf?p=thrift.git"
/>
@@ -176,5 +178,5 @@
<!-- run with: ant -Drelease=true publish -->
<signAndDeploy file="${as3.pom.xml}" packaging="pom" classifier="" pom="${as3.pom.xml}"/>
<signAndDeploy file="${as3.swc.file}" packaging="swc" classifier="" pom="${as3.pom.xml}"/>
</target>
</target>
</project>

40
vendor/git.apache.org/thrift.git/lib/cl/Makefile.am generated vendored Normal file
View File

@@ -0,0 +1,40 @@
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
#
THRIFT = $(top_builddir)/compiler/cpp/thrift
all-local:
bash ensure-externals.sh
run-tests: test/make-test-binary.lisp
$(SBCL) --script test/make-test-binary.lisp
check-local: run-tests
./run-tests
clean-local:
$(RM) run-tests quicklisp.lisp backport-update.zip
$(RM) -rf lib externals quicklisp
EXTRA_DIST = \
README.md \
READMES \
load-locally.lisp \
test \
ensure-externals.sh

253
vendor/git.apache.org/thrift.git/lib/cl/README.md generated vendored Normal file
View File

@@ -0,0 +1,253 @@
Thrift Common Lisp Library
License
=======
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
Using Thrift with Common Lisp
============================
Thrift is a protocol and library for language-independent communication between cooperating
processes. The communication takes the form of request and response messages, of which the forms
are specified in advance throufh a shared interface definition. A Thrift definition file is translated
into Lisp source files, which comprise several definitions:
* Three packages, one for the namespace of the implementation operators, and one each for request and
response operators.
* Various type definitions as implementations for Thrift typedef and enum definitions.
* DEF-STRUCT and DEF-EXCEPTION forms for Thrift struct and exception definitions.
* DEF-SERVICE forms for thrift service definitions.
Each service definition expands in a collection of generic function definitions. For each `op`
in the service definition, two functions are defined
* `op`-request is defined for use by a client. It accepts an additional initial `protocol` argument,
to act as the client proxy for the operation and mediate the interaction with a remote process
through a Thrift-encoded transport stream.
* `op`-response is defined for use by a server. It accepts a single `protocol` argument. A server
uses it to decode the request message, invoke the base `op` function with the message arguments,
encode and send the the result as a response, and handles exceptions.
The client interface is one operator
* `with-client (variable location) . body` : creates a connection in a dynamic context and closes it
upon exit. The variable is bound to a client proxy stream/protocol instance, which wraps the
base i/o stream - socket, file, etc, with an operators which implement the Thrift protocol
and transport mechanisms.
The server interface combines server and service objects
* `serve (location service)` : accepts connections on the designated port and responds to
requests of the service's operations.
Building
--------
The Thrift Common Lisp library is packaged as the ASDF[[1]] system `thrift`.
It depends on the systems
* puri[[2]] : for the thrift uri class
* closer-mop[[3]] : for class metadata
* trivial-utf-8[[4]] : for string codecs
* usocket[[5]] : for the socket transport
* ieee-floats[[6]] : for conversion between ints and floats
* trivial-gray-streams[[7]] : an abstraction layer for gray streams
* alexandria[[8]] : handy utilities
The dependencies are bundled for local builds of tests and tutorial binaries -
it is possible to use those bundles to load the library, too.
In order to build it, register those systems with ASDF and evaluate:
(asdf:load-system :thrift)
This will compile and load the Lisp compiler for Thrift definition files, the
transport and protocol implementations, and the client and server interface
functions. In order to use Thrift in an application, one must also author and/or
load the interface definitions for the remote service.[[9]] If one is implementing a service,
one must also define the actual functions to which Thrift is to act as the proxy
interface. The remainder of this document follows the Thrift tutorial to illustrate how
to perform the steps
* implement the service
* translate the Thrift IDL
* load the Lisp service interfaces
* run a server for the service
* use a client to access the service remotely
Note that, if one is to implement a new service, one will also need to author the
IDL files, as there is no facility to generate them from a service implementation.
Implement the Service
---------------------
The tutorial comprises serveral functions: `add`, `ping`, `zip`, and `calculate`.
Each translated IDL file generates three packages for every service. In the case of
the tutorial file, the relevant packages are:
* tutorial.calculator
* tutorial.calculator-implementation
* tutorial.calculator-response
This is to separate the request (generated), response (generated) and implementation
(meant to be implemented by the programmer) functions for defined Thrift methods.
It is suggested to work in the `tutorial-implementation` package while implementing
the services - it imports the `common-lisp` package, while the service-specific ones
don't (to avoid conflicts between Thrift method names and function names in `common-lisp`).
;; define the base operations
(in-package :tutorial-implementation)
(defun tutorial.calculator-implementation:add (num1 num2)
(format t "~&Asked to add ~A and ~A." num1 num2)
(+ num1 num2))
(defun tutorial.calculator-implementation:ping ()
(print :ping))
(defun tutorial.calculator-implementation:zip ()
(print :zip))
(defun tutorial.calculator-implementation:calculate (logid task)
(calculate-op (work-op task) (work-num1 task) (work-num2 task)))
(defgeneric calculate-op (op arg1 arg2)
(:method :around (op arg1 arg2)
(let ((result (call-next-method)))
(format t "~&Asked to calculate: ~d on ~A and ~A = ~d." op arg1 arg2 result)
result))
(:method ((op (eql operation.add)) arg1 arg2)
(+ arg1 arg2))
(:method ((op (eql operation.subtract)) arg1 arg2)
(- arg1 arg2))
(:method ((op (eql operation.multiply)) arg1 arg2)
(* arg1 arg2))
(:method ((op (eql operation.divide)) arg1 arg2)
(/ arg1 arg2)))
(defun zip () (print 'zip))
Translate the Thrift IDL
------------------------
IDL files employ the file extension `thrift`. In this case, there are two files to translate
* `tutorial.thrift`
* `shared.thrift`
As the former includes the latter, one uses it to generate the interfaces:
$THRIFT/bin/thrift -r --gen cl $THRIFT/tutorial/tutorial.thrift
`-r` stands for recursion, while `--gen` lets one choose the language to translate to.
Load the Lisp translated service interfaces
-------------------------------------------
The translator generates three files for each IDL file. For example `tutorial-types.lisp`,
`tutorial-vars.lisp` and an `.asd` file that can be used to load them both and pull in
other includes (like `shared` within the tutorial) as dependencies.
Run a Server for the Service
----------------------------
The actual service name, as specified in the `def-service` form in `tutorial.lisp`, is `calculator`.
Each service definition defines a global variable with the service name and binds it to a
service instance whch describes the operations.
In order to start a service, specify a location and the service instance.
(in-package :tutorial)
(serve #u"thrift://127.0.0.1:9091" calculator)
Use a Client to Access the Service Remotely
-------------------------------------------
[in some other process] run the client
(in-package :cl-user)
(macrolet ((show (form)
`(format *trace-output* "~%~s =>~{ ~s~}"
',form
(multiple-value-list (ignore-errors ,form)))))
(with-client (protocol #u"thrift://127.0.0.1:9091")
(show (tutorial.calculator:ping protocol))
(show (tutorial.calculator:add protocol 1 2))
(show (tutorial.calculator:add protocol 1 4))
(let ((task (make-instance 'tutorial:work
:op operation.subtract :num1 15 :num2 10)))
(show (tutorial.calculator:calculate protocol 1 task))
(setf (tutorial:work-op task) operation.divide
(tutorial:work-num1 task) 1
(tutorial:work-num2 task) 0)
(show (tutorial.calculator:calculate protocol 1 task)))
(show (shared.shared-service:get-struct protocol 1))
(show (zip protocol))))
Issues
------
### optional fields
Where the IDL declares a field options, the def-struct form includes no
initform for the slot and the encoding operator skips an unbound slot. This leave some ambiguity
with bool fields.
### instantiation protocol :
struct classes are standard classes and exception classes are
whatever the implementation prescribes. decoders apply make-struct to an initargs list.
particularly at the service end, there are advantages to resourcing structs and decoding
with direct side-effects on slot-values
### maps:
Maps are now represented as hash tables. As data through the call/reply interface is all statically
typed, it is not necessary for the objects to themselves indicate the coding form. Association lists
would be sufficient. As the key type is arbitrary, property lists offer no additional convenience:
as `getf` operates with `eq` a new access interface would be necessary and they would not be
available for function application.
[1]: www.common-lisp.net/asdf
[2]: http://github.com/lisp/com.b9.puri.ppcre
[3]: www.common-lisp.net/closer-mop
[4]: trivial-utf-8
[5]: https://github.com/usocket/usocket
[6]: https://github.com/marijnh/ieee-floats
[7]: https://github.com/trivial-gray-streams/trivial-gray-streams
[8]: https://gitlab.common-lisp.net/alexandria/alexandria
[9]: http://wiki.apache.org/thrift/ThriftGeneration
* usocket[[5]] : for the socket transport
* ieee-floats[[6]] : for conversion between ints and floats
* trivial-gray-streams[[7]] : an abstraction layer for gray streams
* alexandria[[8]] : handy utilities

View File

@@ -0,0 +1,64 @@
(in-package :cl-user)
#+(or ccl sbcl) /development/source/library/
(load "build-init.lisp")
;;; ! first, select the api version in the cassandra system definition
;;; as only one should be loaded at a time.
(asdf:load-system :de.setf.cassandra)
(in-package :de.setf.cassandra)
(defparameter *c-location*
;; remote
;; #u"thrift://ec2-174-129-66-148.compute-1.amazonaws.com:9160"
;; local
#u"thrift://127.0.0.1:9160"
"A cassandra service location - either the local one or a remote service
- always a 'thrift' uri.")
(defparameter *c* (thrift:client *c-location*))
(cassandra:describe-keyspaces *c*)
;; => ("Keyspace1" "system")
(cassandra:describe-cluster-name *c*)
;; =>"Test Cluster"
(cassandra:describe-version *c*)
;; => "2.1.0"
(loop for space in (cassandra:describe-keyspaces *c*)
collect (loop for key being each hash-key of (cassandra:describe-keyspace *c* space)
using (hash-value value)
collect (cons key
(loop for key being each hash-key of value
using (hash-value value)
collect (cons key value)))))
(close *c*)
(defun describe-cassandra (location &optional (stream *standard-output*))
"Print the first-order store metadata for a cassandra LOCATION."
(thrift:with-client (cassandra location)
(let* ((keyspace-names (cassandra:describe-keyspaces cassandra))
(cluster (cassandra:describe-cluster-name cassandra))
(version (cassandra:describe-version cassandra))
(keyspace-descriptions (loop for space in keyspace-names
collect (cons space
(loop for key being each hash-key
of (cassandra:describe-keyspace cassandra space)
using (hash-value value)
collect (cons key
(loop for key being each hash-key of value
using (hash-value value)
collect (cons key value))))))))
(format stream "~&connection to : ~a" cassandra)
(format stream "~&version : ~a" version)
(format stream "~&cluster : ~a" cluster)
(format stream "~&keyspaces~{~{~%~%space: ~a~@{~% ~{~a :~@{~20t~:w~^~%~}~}~}~}~}" keyspace-descriptions))))
;;; (describe-cassandra *c-location*)

16
vendor/git.apache.org/thrift.git/lib/cl/ensure-externals.sh generated vendored Executable file
View File

@@ -0,0 +1,16 @@
#!/bin/bash
set -e
if [[ ! -e quicklisp.lisp ]]; then curl -O https://beta.quicklisp.org/quicklisp.lisp; fi
sbcl --load quicklisp.lisp \
--eval "(ignore-errors (quicklisp-quickstart:install :path \"quicklisp/\"))" \
--eval "(load \"quicklisp/setup.lisp\")" \
--eval "(quicklisp:bundle-systems '(#:puri #:usocket #:closer-mop #:trivial-utf-8 #:ieee-floats #:trivial-gray-streams #:alexandria #:bordeaux-threads #:cl-ppcre #:fiasco #:net.didierverna.clon) :to \"externals/\")" \
--eval "(quit)" \
--no-userinit
if [[ ! -e backport-update.zip ]]; then
curl -O -L https://github.com/TurtleWarePL/de.setf.thrift/archive/backport-update.zip;
fi
mkdir -p lib
unzip -u backport-update.zip -d lib

View File

@@ -0,0 +1,23 @@
(in-package #:cl-user)
;;;; Licensed under the Apache License, Version 2.0 (the "License");
;;;; you may not use this file except in compliance with the License.
;;;; You may obtain a copy of the License at
;;;;
;;;; http://www.apache.org/licenses/LICENSE-2.0
;;;;
;;;; Unless required by applicable law or agreed to in writing, software
;;;; distributed under the License is distributed on an "AS IS" BASIS,
;;;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;;;; See the License for the specific language governing permissions and
;;;; limitations under the License.
;;;; Just a script for loading the library itself, using bundled dependencies.
;;;; This is here for when we want to build the self-test and cross-test
;;;; binaries.
(require "asdf")
(load (merge-pathnames "externals/bundle.lisp" *load-truename*))
(asdf:load-asd (merge-pathnames "lib/de.setf.thrift-backport-update/thrift.asd" *load-truename*))
(asdf:load-system :thrift)

View File

@@ -0,0 +1,31 @@
;;;; Licensed under the Apache License, Version 2.0 (the "License");
;;;; you may not use this file except in compliance with the License.
;;;; You may obtain a copy of the License at
;;;;
;;;; http://www.apache.org/licenses/LICENSE-2.0
;;;;
;;;; Unless required by applicable law or agreed to in writing, software
;;;; distributed under the License is distributed on an "AS IS" BASIS,
;;;; WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
;;;; See the License for the specific language governing permissions and
;;;; limitations under the License.
;;;; This file is used to build the binary that runs all self-tests. The
;;;; binary is then meant to be hooked up to Thrift's `make check` facility,
;;;; but can easily be run on its own as well.
(in-package #:cl-user)
(require "asdf")
(load (merge-pathnames "../load-locally.lisp" *load-truename*))
(asdf:load-asd (merge-pathnames "../lib/de.setf.thrift-backport-update/test/thrift-test.asd" *load-truename*))
(asdf:load-system :thrift-test)
(asdf:load-system :net.didierverna.clon)
(net.didierverna.clon:nickname-package)
(defun main ()
(let ((result (if (fiasco:run-tests 'thrift-test) 0 -1)))
(clon:exit result)))
(clon:dump "run-tests" main)

View File

@@ -17,4 +17,4 @@
* under the License.
*/
#define ThriftVersion @"0.11.0"
#define ThriftVersion @"1.0.0-dev"

View File

@@ -26,14 +26,14 @@
/**
* De-serialize object from the given input protocol
*
* @param input protocol used for reading
* @param inProtocol protocol used for reading
*/
-(BOOL) read:(id <TProtocol>)inProtocol error:(NSError **)error;
/**
* Serialize object to the given protocol
*
* @param buf output protocol used for writing
* @param outProtocol output protocol used for writing
*/
-(BOOL) write:(id <TProtocol>)outProtocol error:(NSError **)error;

View File

@@ -150,7 +150,7 @@ libthriftz_la_CXXFLAGS = $(AM_CXXFLAGS)
libthriftqt_la_CXXFLAGS = $(AM_CXXFLAGS)
libthriftqt5_la_CXXFLAGS = $(AM_CXXFLAGS)
libthriftnb_la_LDFLAGS = -release $(VERSION) $(BOOST_LDFLAGS)
libthriftz_la_LDFLAGS = -release $(VERSION) $(BOOST_LDFLAGS)
libthriftz_la_LDFLAGS = -release $(VERSION) $(BOOST_LDFLAGS) $(ZLIB_LIBS)
libthriftqt_la_LDFLAGS = -release $(VERSION) $(BOOST_LDFLAGS) $(QT_LIBS)
libthriftqt5_la_LDFLAGS = -release $(VERSION) $(BOOST_LDFLAGS) $(QT5_LIBS)

View File

@@ -94,7 +94,7 @@ void TOutput::errorTimeWrapper(const char* msg) {
}
void TOutput::perror(const char* message, int errno_copy) {
std::string out = message + strerror_s(errno_copy);
std::string out = message + std::string(": ") + strerror_s(errno_copy);
f_(out.c_str());
}

View File

@@ -38,13 +38,14 @@ TEvhttpClientChannel::TEvhttpClientChannel(const std::string& host,
const std::string& path,
const char* address,
int port,
struct event_base* eb)
struct event_base* eb,
struct evdns_base* dnsbase)
: host_(host), path_(path), conn_(NULL) {
conn_ = evhttp_connection_new(address, port);
conn_ = evhttp_connection_base_new(eb, dnsbase, address, port);
if (conn_ == NULL) {
throw TException("evhttp_connection_new failed");
}
evhttp_connection_set_base(conn_, eb);
}
TEvhttpClientChannel::~TEvhttpClientChannel() {

View File

@@ -27,6 +27,7 @@
#include <thrift/async/TAsyncChannel.h>
struct event_base;
struct evdns_base;
struct evhttp_connection;
struct evhttp_request;
@@ -50,7 +51,8 @@ public:
const std::string& path,
const char* address,
int port,
struct event_base* eb);
struct event_base* eb,
struct evdns_base *dnsbase = 0);
~TEvhttpClientChannel();
virtual void sendAndRecvMessage(const VoidCallback& cob,

View File

@@ -472,6 +472,18 @@ void TNonblockingServer::TConnection::workSocket() {
}
// size known; now get the rest of the frame
transition();
// If the socket has more data than the frame header, continue to work on it. This is not strictly necessary for
// regular sockets, because if there is more data, libevent will fire the event handler registered for read
// readiness, which will in turn call workSocket(). However, some socket types (such as TSSLSocket) may have the
// data sitting in their internal buffers and from libevent's perspective, there is no further data available. In
// that case, not having this workSocket() call here would result in a hang as we will never get to work the socket,
// despite having more data.
if (tSocket_->hasPendingDataToRead())
{
workSocket();
}
return;
case SOCKET_RECV:
@@ -647,6 +659,7 @@ void TNonblockingServer::TConnection::transition() {
return;
}
}
// fallthrough
// Intentionally fall through here, the call to process has written into
// the writeBuffer_
@@ -677,9 +690,6 @@ void TNonblockingServer::TConnection::transition() {
appState_ = APP_SEND_RESULT;
setWrite();
// Try to work the socket immediately
// workSocket();
return;
}
@@ -698,6 +708,7 @@ void TNonblockingServer::TConnection::transition() {
server_->getIdleWriteBufferLimit());
callsForResize_ = 0;
}
// fallthrough
// N.B.: We also intentionally fall through here into the INIT state!
@@ -718,9 +729,6 @@ void TNonblockingServer::TConnection::transition() {
// Register read event
setRead();
// Try to work the socket right away
// workSocket();
return;
case APP_READ_FRAME_SIZE:
@@ -753,9 +761,6 @@ void TNonblockingServer::TConnection::transition() {
socketState_ = SOCKET_RECV;
appState_ = APP_READ_REQUEST;
// Work the socket right away
workSocket();
return;
case APP_CLOSE_CONNECTION:
@@ -1063,7 +1068,7 @@ void TNonblockingServer::expireClose(stdcxx::shared_ptr<Runnable> task) {
connection->forceClose();
}
void TNonblockingServer::stop() {
void TNonblockingServer::stop() {
// Breaks the event loop in all threads so that they end ASAP.
for (uint32_t i = 0; i < ioThreads_.size(); ++i) {
ioThreads_[i]->stop();

View File

@@ -51,6 +51,8 @@
# define THRIFT_LSEEK _lseek
# define THRIFT_WRITE _write
# define THRIFT_READ _read
# define THRIFT_IOCTL_SOCKET ioctlsocket
# define THRIFT_IOCTL_SOCKET_NUM_BYTES_TYPE u_long
# define THRIFT_FSTAT _fstat
# define THRIFT_STAT _stat
# ifdef _WIN32_WCE
@@ -111,6 +113,8 @@
# define THRIFT_LSEEK lseek
# define THRIFT_WRITE write
# define THRIFT_READ read
# define THRIFT_IOCTL_SOCKET ioctl
# define THRIFT_IOCTL_SOCKET_NUM_BYTES_TYPE int
# define THRIFT_STAT stat
# define THRIFT_FSTAT fstat
# define THRIFT_GAI_STRERROR gai_strerror

View File

@@ -368,7 +368,7 @@ void TMemoryBuffer::ensureCanWrite(uint32_t len) {
throw TTransportException(TTransportException::BAD_ARGS,
"Internal buffer size overflow");
}
avail = available_write() + (new_size - bufferSize_);
avail = available_write() + (static_cast<uint32_t>(new_size) - bufferSize_);
}
// Allocate into a new pointer so we don't bork ours if it fails.
@@ -382,7 +382,7 @@ void TMemoryBuffer::ensureCanWrite(uint32_t len) {
wBase_ = new_buffer + (wBase_ - buffer_);
wBound_ = new_buffer + new_size;
buffer_ = new_buffer;
bufferSize_ = new_size;
bufferSize_ = static_cast<uint32_t>(new_size);
}
void TMemoryBuffer::writeSlow(const uint8_t* buf, uint32_t len) {

View File

@@ -84,8 +84,10 @@ uint32_t THttpTransport::readEnd() {
uint32_t THttpTransport::readMoreData() {
uint32_t size;
// Get more data!
refill();
if (httpPos_ == httpBufLen_) {
// Get more data!
refill();
}
if (readHeaders_) {
readHeaders();

View File

@@ -249,6 +249,17 @@ TSSLSocket::~TSSLSocket() {
close();
}
bool TSSLSocket::hasPendingDataToRead() {
if (!isOpen()) {
return false;
}
initializeHandshake();
if (!checkHandshake())
throw TSSLException("TSSLSocket::hasPendingDataToRead: Handshake is not completed");
// data may be available in SSL buffers (note: SSL_pending does not have a failure mode)
return SSL_pending(ssl_) > 0 || TSocket::hasPendingDataToRead();
}
void TSSLSocket::init() {
handshakeCompleted_ = false;
readRetryCount_ = 0;
@@ -293,6 +304,7 @@ bool TSSLSocket::peek() {
&& (errno_copy != THRIFT_EAGAIN)) {
break;
}
// fallthrough
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
// in the case of SSL_ERROR_SYSCALL we want to wait for an read event again
@@ -339,6 +351,7 @@ void TSSLSocket::close() {
&& (errno_copy != THRIFT_EAGAIN)) {
break;
}
// fallthrough
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
// in the case of SSL_ERROR_SYSCALL we want to wait for an write/read event again
@@ -404,6 +417,8 @@ uint32_t TSSLSocket::read(uint8_t* buf, uint32_t len) {
// a certain number
break;
}
// fallthrough
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
if (isLibeventSafe()) {
@@ -460,6 +475,7 @@ void TSSLSocket::write(const uint8_t* buf, uint32_t len) {
&& (errno_copy != THRIFT_EAGAIN)) {
break;
}
// fallthrough
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
if (isLibeventSafe()) {
@@ -504,6 +520,7 @@ uint32_t TSSLSocket::write_partial(const uint8_t* buf, uint32_t len) {
&& (errno_copy != THRIFT_EAGAIN)) {
break;
}
// fallthrough
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
if (isLibeventSafe()) {
@@ -591,6 +608,7 @@ void TSSLSocket::initializeHandshake() {
&& (errno_copy != THRIFT_EAGAIN)) {
break;
}
// fallthrough
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
if (isLibeventSafe()) {
@@ -623,6 +641,7 @@ void TSSLSocket::initializeHandshake() {
&& (errno_copy != THRIFT_EAGAIN)) {
break;
}
// fallthrough
case SSL_ERROR_WANT_READ:
case SSL_ERROR_WANT_WRITE:
if (isLibeventSafe()) {

View File

@@ -78,6 +78,7 @@ public:
bool peek();
void open();
void close();
bool hasPendingDataToRead();
uint32_t read(uint8_t* buf, uint32_t len);
void write(const uint8_t* buf, uint32_t len);
uint32_t write_partial(const uint8_t* buf, uint32_t len);

View File

@@ -21,6 +21,9 @@
#include <cstring>
#include <sstream>
#ifdef HAVE_SYS_IOCTL_H
#include <sys/ioctl.h>
#endif
#ifdef HAVE_SYS_SOCKET_H
#include <sys/socket.h>
#endif
@@ -167,6 +170,26 @@ TSocket::~TSocket() {
close();
}
bool TSocket::hasPendingDataToRead() {
if (!isOpen()) {
return false;
}
int32_t retries = 0;
THRIFT_IOCTL_SOCKET_NUM_BYTES_TYPE numBytesAvailable;
try_again:
int r = THRIFT_IOCTL_SOCKET(socket_, FIONREAD, &numBytesAvailable);
if (r == -1) {
int errno_copy = THRIFT_GET_SOCKET_ERROR;
if (errno_copy == THRIFT_EINTR && (retries++ < maxRecvRetries_)) {
goto try_again;
}
GlobalOutput.perror("TSocket::hasPendingDataToRead() THRIFT_IOCTL_SOCKET() " + getSocketInfo(), errno_copy);
throw TTransportException(TTransportException::UNKNOWN, "Unknown", errno_copy);
}
return numBytesAvailable > 0;
}
bool TSocket::isOpen() {
return (socket_ != THRIFT_INVALID_SOCKET);
}
@@ -786,11 +809,15 @@ void TSocket::setMaxRecvRetries(int maxRecvRetries) {
string TSocket::getSocketInfo() {
std::ostringstream oss;
if (host_.empty() || port_ == 0) {
oss << "<Host: " << getPeerAddress();
oss << " Port: " << getPeerPort() << ">";
if (path_.empty()) {
if (host_.empty() || port_ == 0) {
oss << "<Host: " << getPeerAddress();
oss << " Port: " << getPeerPort() << ">";
} else {
oss << "<Host: " << host_ << " Port: " << port_ << ">";
}
} else {
oss << "<Host: " << host_ << " Port: " << port_ << ">";
oss << "<Path: " << path_ << ">";
}
return oss.str();
}

View File

@@ -84,7 +84,9 @@ public:
virtual bool isOpen();
/**
* Calls select on the socket to see if there is more data available.
* Checks whether there is more data available in the socket to read.
*
* This call blocks until at least one byte is available or the socket is closed.
*/
virtual bool peek();
@@ -100,6 +102,17 @@ public:
*/
virtual void close();
/**
* Determines whether there is pending data to read or not.
*
* This call does not block.
* \throws TTransportException of types:
* NOT_OPEN means the socket has been closed
* UNKNOWN means something unexpected happened
* \returns true if there is pending data to read, false otherwise
*/
virtual bool hasPendingDataToRead();
/**
* Reads from the underlying socket.
* \returns the number of bytes read or 0 indicates EOF

View File

@@ -70,7 +70,6 @@
#pragma warning(disable : 4996) // Deprecated posix name.
#endif
#define VERSION "0.11.0"
#define HAVE_GETTIMEOFDAY 1
#define HAVE_SYS_STAT_H 1

View File

@@ -1,4 +1,4 @@
#
#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
@@ -19,9 +19,6 @@
include_directories(SYSTEM "${Boost_INCLUDE_DIRS}")
add_definitions("-D__STDC_FORMAT_MACROS")
add_definitions("-D__STDC_LIMIT_MACROS")
if (WITH_DYN_LINK_TEST)
add_definitions( -DBOOST_TEST_DYN_LINK )
endif()
@@ -43,6 +40,10 @@ set(testgencpp_SOURCES
gen-cpp/Recursive_types.h
gen-cpp/ThriftTest_types.cpp
gen-cpp/ThriftTest_types.h
gen-cpp/OneWayTest_types.cpp
gen-cpp/OneWayTest_types.h
gen-cpp/OneWayService.cpp
gen-cpp/OneWayService.h
gen-cpp/TypedefTest_types.cpp
gen-cpp/TypedefTest_types.h
ThriftTest_extras.cpp
@@ -71,6 +72,7 @@ target_link_libraries(Benchmark testgencpp)
set(UnitTest_SOURCES
UnitTestMain.cpp
OneWayHTTPTest.cpp
TMemoryBufferTest.cpp
TBufferBaseTest.cpp
Base64Test.cpp
@@ -394,6 +396,10 @@ add_custom_command(OUTPUT gen-cpp/SecondService.cpp gen-cpp/ThriftTest_constants
COMMAND ${THRIFT_COMPILER} --gen cpp ${PROJECT_SOURCE_DIR}/test/ThriftTest.thrift
)
add_custom_command(OUTPUT gen-cpp/OneWayService.cpp gen-cpp/OneWayTest_constants.cpp gen-cpp/OneWayTest_types.h gen-cpp/OneWayService.h gen-cpp/OneWayTest_constants.h gen-cpp/OneWayTest_types.cpp
COMMAND ${THRIFT_COMPILER} --gen cpp ${CMAKE_CURRENT_SOURCE_DIR}/OneWayTest.thrift
)
add_custom_command(OUTPUT gen-cpp/ChildService.cpp gen-cpp/ChildService.h gen-cpp/ParentService.cpp gen-cpp/ParentService.h gen-cpp/proc_types.cpp gen-cpp/proc_types.h
COMMAND ${THRIFT_COMPILER} --gen cpp:templates,cob_style ${CMAKE_CURRENT_SOURCE_DIR}/processor/proc.thrift
)

View File

@@ -28,6 +28,9 @@ BUILT_SOURCES = gen-cpp/AnnotationTest_types.h \
gen-cpp/ChildService.h \
gen-cpp/EmptyService.h \
gen-cpp/ParentService.h \
gen-cpp/OneWayTest_types.h \
gen-cpp/OneWayService.h \
gen-cpp/OneWayTest_constants.h \
gen-cpp/proc_types.h
noinst_LTLIBRARIES = libtestgencpp.la libprocessortest.la
@@ -36,6 +39,8 @@ nodist_libtestgencpp_la_SOURCES = \
gen-cpp/AnnotationTest_types.h \
gen-cpp/DebugProtoTest_types.cpp \
gen-cpp/DebugProtoTest_types.h \
gen-cpp/DoubleConstantsTest_constants.cpp \
gen-cpp/DoubleConstantsTest_constants.h \
gen-cpp/EnumTest_types.cpp \
gen-cpp/EnumTest_types.h \
gen-cpp/OptionalRequiredTest_types.cpp \
@@ -48,6 +53,12 @@ nodist_libtestgencpp_la_SOURCES = \
gen-cpp/ThriftTest_constants.h \
gen-cpp/TypedefTest_types.cpp \
gen-cpp/TypedefTest_types.h \
gen-cpp/OneWayService.cpp \
gen-cpp/OneWayTest_constants.cpp \
gen-cpp/OneWayTest_types.h \
gen-cpp/OneWayService.h \
gen-cpp/OneWayTest_constants.h \
gen-cpp/OneWayTest_types.cpp \
ThriftTest_extras.cpp \
DebugProtoTest_extras.cpp
@@ -93,6 +104,7 @@ check_PROGRAMS = \
link_test \
OpenSSLManualInitTest \
EnumTest \
RenderedDoubleConstantsTest \
AnnotationTest
if AMX_HAVE_LIBEVENT
@@ -113,6 +125,7 @@ TESTS = \
UnitTests_SOURCES = \
UnitTestMain.cpp \
OneWayHTTPTest.cpp \
TMemoryBufferTest.cpp \
TBufferBaseTest.cpp \
Base64Test.cpp \
@@ -130,7 +143,9 @@ endif
UnitTests_LDADD = \
libtestgencpp.la \
$(BOOST_TEST_LDADD)
$(BOOST_TEST_LDADD) \
$(BOOST_SYSTEM_LDADD) \
$(BOOST_THREAD_LDADD)
TInterruptTest_SOURCES = \
TSocketInterruptTest.cpp \
@@ -190,6 +205,10 @@ EnumTest_LDADD = \
libtestgencpp.la \
$(BOOST_TEST_LDADD)
RenderedDoubleConstantsTest_SOURCES = RenderedDoubleConstantsTest.cpp
RenderedDoubleConstantsTest_LDADD = libtestgencpp.la $(BOOST_TEST_LDADD)
AnnotationTest_SOURCES = \
AnnotationTest.cpp
@@ -367,6 +386,10 @@ gen-cpp/AnnotationTest_constants.cpp gen-cpp/AnnotationTest_constants.h gen-cpp/
gen-cpp/DebugProtoTest_types.cpp gen-cpp/DebugProtoTest_types.h gen-cpp/EmptyService.cpp gen-cpp/EmptyService.h: $(top_srcdir)/test/DebugProtoTest.thrift
$(THRIFT) --gen cpp $<
gen-cpp/DoubleConstantsTest_constants.cpp gen-cpp/DoubleConstantsTest_constants.h: $(top_srcdir)/test/DoubleConstantsTest.thrift
$(THRIFT) --gen cpp $<
gen-cpp/EnumTest_types.cpp gen-cpp/EnumTest_types.h: $(top_srcdir)/test/EnumTest.thrift
$(THRIFT) --gen cpp $<
@@ -385,6 +408,9 @@ gen-cpp/Service.cpp gen-cpp/StressTest_types.cpp: $(top_srcdir)/test/StressTest.
gen-cpp/SecondService.cpp gen-cpp/ThriftTest_constants.cpp gen-cpp/ThriftTest.cpp gen-cpp/ThriftTest_types.cpp gen-cpp/ThriftTest_types.h: $(top_srcdir)/test/ThriftTest.thrift
$(THRIFT) --gen cpp $<
gen-cpp/OneWayService.cpp gen-cpp/OneWayTest_constants.cpp gen-cpp/OneWayTest_types.h gen-cpp/OneWayService.h gen-cpp/OneWayTest_constants.h gen-cpp/OneWayTest_types.cpp: OneWayTest.thrift
$(THRIFT) --gen cpp $<
gen-cpp/ChildService.cpp gen-cpp/ChildService.h gen-cpp/ParentService.cpp gen-cpp/ParentService.h gen-cpp/proc_types.cpp gen-cpp/proc_types.h: processor/proc.thrift
$(THRIFT) --gen cpp:templates,cob_style $<
@@ -401,4 +427,5 @@ EXTRA_DIST = \
qt \
CMakeLists.txt \
DebugProtoTest_extras.cpp \
ThriftTest_extras.cpp
ThriftTest_extras.cpp \
OneWayTest.thrift

View File

@@ -0,0 +1,242 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
#include <boost/test/auto_unit_test.hpp>
#include <boost/thread.hpp>
#include <iostream>
#include <climits>
#include <vector>
#include <thrift/concurrency/Monitor.h>
#include <thrift/protocol/TBinaryProtocol.h>
#include <thrift/protocol/TJSONProtocol.h>
#include <thrift/server/TThreadedServer.h>
#include <thrift/transport/THttpServer.h>
#include <thrift/transport/THttpClient.h>
#include <thrift/transport/TServerSocket.h>
#include <thrift/transport/TSocket.h>
#include <thrift/stdcxx.h>
#include <thrift/transport/TBufferTransports.h>
#include "gen-cpp/OneWayService.h"
BOOST_AUTO_TEST_SUITE(OneWayHTTPTest)
using namespace apache::thrift;
using apache::thrift::protocol::TProtocol;
using apache::thrift::protocol::TBinaryProtocol;
using apache::thrift::protocol::TBinaryProtocolFactory;
using apache::thrift::protocol::TJSONProtocol;
using apache::thrift::protocol::TJSONProtocolFactory;
using apache::thrift::server::TThreadedServer;
using apache::thrift::server::TServerEventHandler;
using apache::thrift::transport::TTransport;
using apache::thrift::transport::THttpServer;
using apache::thrift::transport::THttpServerTransportFactory;
using apache::thrift::transport::THttpClient;
using apache::thrift::transport::TBufferedTransport;
using apache::thrift::transport::TBufferedTransportFactory;
using apache::thrift::transport::TMemoryBuffer;
using apache::thrift::transport::TServerSocket;
using apache::thrift::transport::TSocket;
using apache::thrift::transport::TTransportException;
using apache::thrift::stdcxx::shared_ptr;
using std::cout;
using std::cerr;
using std::endl;
using std::string;
namespace utf = boost::unit_test;
// Define this env var to enable some logging (in case you need to debug)
#undef ENABLE_STDERR_LOGGING
class OneWayServiceHandler : public onewaytest::OneWayServiceIf {
public:
OneWayServiceHandler() {}
void roundTripRPC() override {
#ifdef ENABLE_STDERR_LOGGING
cerr << "roundTripRPC()" << endl;
#endif
}
void oneWayRPC() {
#ifdef ENABLE_STDERR_LOGGING
cerr << "oneWayRPC()" << std::endl ;
#endif
}
};
class OneWayServiceCloneFactory : virtual public onewaytest::OneWayServiceIfFactory {
public:
virtual ~OneWayServiceCloneFactory() {}
virtual onewaytest::OneWayServiceIf* getHandler(const ::apache::thrift::TConnectionInfo& connInfo)
{
(void)connInfo ;
return new OneWayServiceHandler;
}
virtual void releaseHandler( onewaytest::OneWayServiceIf* handler) {
delete handler;
}
};
class RPC0ThreadClass {
public:
RPC0ThreadClass(TThreadedServer& server) : server_(server) { } // Constructor
~RPC0ThreadClass() { } // Destructor
void Run() {
server_.serve() ;
}
TThreadedServer& server_ ;
} ;
using apache::thrift::concurrency::Monitor;
using apache::thrift::concurrency::Mutex;
using apache::thrift::concurrency::Synchronized;
// copied from IntegrationTest
class TServerReadyEventHandler : public TServerEventHandler, public Monitor {
public:
TServerReadyEventHandler() : isListening_(false), accepted_(0) {}
virtual ~TServerReadyEventHandler() {}
virtual void preServe() {
Synchronized sync(*this);
isListening_ = true;
notify();
}
virtual void* createContext(shared_ptr<TProtocol> input,
shared_ptr<TProtocol> output) {
Synchronized sync(*this);
++accepted_;
notify();
(void)input;
(void)output;
return NULL;
}
bool isListening() const { return isListening_; }
uint64_t acceptedCount() const { return accepted_; }
private:
bool isListening_;
uint64_t accepted_;
};
class TBlockableBufferedTransport : public TBufferedTransport {
public:
TBlockableBufferedTransport(stdcxx::shared_ptr<TTransport> transport)
: TBufferedTransport(transport, 10240),
blocked_(false) {
}
uint32_t write_buffer_length() {
uint32_t have_bytes = static_cast<uint32_t>(wBase_ - wBuf_.get());
return have_bytes ;
}
void block() {
blocked_ = true ;
#ifdef ENABLE_STDERR_LOGGING
cerr << "block flushing\n" ;
#endif
}
void unblock() {
blocked_ = false ;
#ifdef ENABLE_STDERR_LOGGING
cerr << "unblock flushing, buffer is\n<<" << std::string((char *)wBuf_.get(), write_buffer_length()) << ">>\n" ;
#endif
}
void flush() override {
if (blocked_) {
#ifdef ENABLE_STDERR_LOGGING
cerr << "flush was blocked\n" ;
#endif
return ;
}
TBufferedTransport::flush() ;
}
bool blocked_ ;
} ;
BOOST_AUTO_TEST_CASE( JSON_BufferedHTTP )
{
stdcxx::shared_ptr<TServerSocket> ss = stdcxx::make_shared<TServerSocket>(0) ;
TThreadedServer server(
stdcxx::make_shared<onewaytest::OneWayServiceProcessorFactory>(stdcxx::make_shared<OneWayServiceCloneFactory>()),
ss, //port
stdcxx::make_shared<THttpServerTransportFactory>(),
stdcxx::make_shared<TJSONProtocolFactory>());
stdcxx::shared_ptr<TServerReadyEventHandler> pEventHandler(new TServerReadyEventHandler) ;
server.setServerEventHandler(pEventHandler);
#ifdef ENABLE_STDERR_LOGGING
cerr << "Starting the server...\n";
#endif
RPC0ThreadClass t(server) ;
boost::thread thread(&RPC0ThreadClass::Run, &t);
{
Synchronized sync(*(pEventHandler.get()));
while (!pEventHandler->isListening()) {
pEventHandler->wait();
}
}
int port = ss->getPort() ;
#ifdef ENABLE_STDERR_LOGGING
cerr << "port " << port << endl ;
#endif
{
stdcxx::shared_ptr<TSocket> socket(new TSocket("localhost", port));
socket->setRecvTimeout(10000) ; // 1000msec should be enough
stdcxx::shared_ptr<TBlockableBufferedTransport> blockable_transport(new TBlockableBufferedTransport(socket));
stdcxx::shared_ptr<TTransport> transport(new THttpClient(blockable_transport, "localhost", "/service"));
stdcxx::shared_ptr<TProtocol> protocol(new TJSONProtocol(transport));
onewaytest::OneWayServiceClient client(protocol);
transport->open();
client.roundTripRPC();
blockable_transport->block() ;
uint32_t size0 = blockable_transport->write_buffer_length() ;
client.send_oneWayRPC() ;
uint32_t size1 = blockable_transport->write_buffer_length() ;
client.send_oneWayRPC() ;
uint32_t size2 = blockable_transport->write_buffer_length() ;
BOOST_CHECK((size1 - size0) == (size2 - size1)) ;
blockable_transport->unblock() ;
client.send_roundTripRPC() ;
blockable_transport->flush() ;
try {
client.recv_roundTripRPC() ;
} catch (TTransportException e) {
BOOST_ERROR( "we should not get a transport exception -- this means we failed: " + std::string(e.what()) ) ;
}
transport->close();
}
server.stop();
thread.join() ;
#ifdef ENABLE_STDERR_LOGGING
cerr << "finished.\n";
#endif
}
BOOST_AUTO_TEST_SUITE_END()

View File

@@ -1,5 +1,3 @@
// +build !go1.7
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -17,32 +15,32 @@
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*
* Contains some contributions under the Thrift Software License.
* Please see doc/old-thrift-license.txt in the Thrift distribution for
* details.
*/
package tests
namespace c_glib OneWayTest
namespace java onewaytest
namespace cpp onewaytest
namespace rb Onewaytest
namespace perl OneWayTest
namespace csharp Onewaytest
namespace js OneWayTest
namespace st OneWayTest
namespace py OneWayTest
namespace py.twisted OneWayTest
namespace go onewaytest
namespace php OneWayTest
namespace delphi Onewaytest
namespace cocoa OneWayTest
namespace lua OneWayTest
namespace xsd test (uri = 'http://thrift.apache.org/ns/OneWayTest')
namespace netcore ThriftAsync.OneWayTest
import (
"fmt"
"golang.org/x/net/context"
)
var defaultCtx = context.Background()
type FirstImpl struct{}
func (f *FirstImpl) ReturnOne(ctx context.Context) (r int64, err error) {
return 1, nil
// a minimal Thrift service, for use in OneWayHTTPTtest.cpp
service OneWayService {
void roundTripRPC(),
oneway void oneWayRPC()
}
type SecondImpl struct{}
func (s *SecondImpl) ReturnTwo(ctx context.Context) (r int64, err error) {
return 2, nil
}
type impl struct{}
func (i *impl) Hi(ctx context.Context, in int64, s string) (err error) { fmt.Println("Hi!"); return }
func (i *impl) Emptyfunc(ctx context.Context) (err error) { return }
func (i *impl) EchoInt(ctx context.Context, param int64) (r int64, err error) { return param, nil }

View File

@@ -0,0 +1,122 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
#define EPSILON 0.0000001
#include <typeindex>
#include <typeinfo>
#include <vector>
#include "gen-cpp/DoubleConstantsTest_constants.h"
using namespace thrift::test;
#define BOOST_TEST_MODULE RenderedDoubleConstantsTest
#include <boost/test/unit_test.hpp>
BOOST_AUTO_TEST_SUITE(RenderedDoubleConstantsTest)
BOOST_AUTO_TEST_CASE(test_rendered_double_constants) {
const double EXPECTED_DOUBLE_ASSIGNED_TO_INT_CONSTANT = 1.0;
const double EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_INT_CONSTANT = -100.0;
const double EXPECTED_DOUBLE_ASSIGNED_TO_LARGEST_INT_CONSTANT = 9223372036854775807.0;
const double EXPECTED_DOUBLE_ASSIGNED_TO_SMALLEST_INT_CONSTANT = -9223372036854775807.0;
const double EXPECTED_DOUBLE_ASSIGNED_TO_DOUBLE_WITH_MANY_DECIMALS = 3.14159265359;
const double EXPECTED_DOUBLE_ASSIGNED_TO_FRACTIONAL_DOUBLE = 1000000.1;
const double EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_FRACTIONAL_DOUBLE = -1000000.1;
const double EXPECTED_DOUBLE_ASSIGNED_TO_LARGE_DOUBLE = 1.7e+308;
const double EXPECTED_DOUBLE_ASSIGNED_TO_LARGE_FRACTIONAL_DOUBLE = 9223372036854775816.43;
const double EXPECTED_DOUBLE_ASSIGNED_TO_SMALL_DOUBLE = -1.7e+308;
const double EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_BUT_LARGE_FRACTIONAL_DOUBLE = -9223372036854775816.43;
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_INT_CONSTANT_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_INT_CONSTANT, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_NEGATIVE_INT_CONSTANT_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_INT_CONSTANT, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_LARGEST_INT_CONSTANT_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_LARGEST_INT_CONSTANT, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_SMALLEST_INT_CONSTANT_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_SMALLEST_INT_CONSTANT, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_DOUBLE_WITH_MANY_DECIMALS_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_DOUBLE_WITH_MANY_DECIMALS, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_FRACTIONAL_DOUBLE_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_FRACTIONAL_DOUBLE, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_NEGATIVE_FRACTIONAL_DOUBLE_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_FRACTIONAL_DOUBLE, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_LARGE_DOUBLE_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_LARGE_DOUBLE, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_LARGE_FRACTIONAL_DOUBLE_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_LARGE_FRACTIONAL_DOUBLE, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_SMALL_DOUBLE_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_SMALL_DOUBLE, EPSILON);
BOOST_CHECK_CLOSE(
g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_NEGATIVE_BUT_LARGE_FRACTIONAL_DOUBLE_TEST,
EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_BUT_LARGE_FRACTIONAL_DOUBLE, EPSILON);
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_INT_CONSTANT_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_INT_CONSTANT).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_NEGATIVE_INT_CONSTANT_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_INT_CONSTANT).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_LARGEST_INT_CONSTANT_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_LARGEST_INT_CONSTANT).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_SMALLEST_INT_CONSTANT_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_SMALLEST_INT_CONSTANT).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_DOUBLE_WITH_MANY_DECIMALS_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_DOUBLE_WITH_MANY_DECIMALS).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_FRACTIONAL_DOUBLE_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_FRACTIONAL_DOUBLE).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_NEGATIVE_FRACTIONAL_DOUBLE_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_FRACTIONAL_DOUBLE).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_LARGE_DOUBLE_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_LARGE_DOUBLE).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_LARGE_FRACTIONAL_DOUBLE_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_LARGE_FRACTIONAL_DOUBLE).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_SMALL_DOUBLE_TEST).hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_SMALL_DOUBLE).hash_code());
BOOST_CHECK(
typeid(g_DoubleConstantsTest_constants.DOUBLE_ASSIGNED_TO_NEGATIVE_BUT_LARGE_FRACTIONAL_DOUBLE_TEST)
.hash_code() ==
typeid(EXPECTED_DOUBLE_ASSIGNED_TO_NEGATIVE_BUT_LARGE_FRACTIONAL_DOUBLE).hash_code());
}
BOOST_AUTO_TEST_CASE(test_rendered_double_list) {
const std::vector<double> EXPECTED_DOUBLE_LIST{1.0,-100.0,100.0,9223372036854775807.0,-9223372036854775807.0,
3.14159265359,1000000.1,-1000000.1,1.7e+308,-1.7e+308,9223372036854775816.43,-9223372036854775816.43};
BOOST_CHECK_EQUAL(g_DoubleConstantsTest_constants.DOUBLE_LIST_TEST.size(), EXPECTED_DOUBLE_LIST.size());
for (unsigned int i = 0; i < EXPECTED_DOUBLE_LIST.size(); ++i) {
BOOST_CHECK_CLOSE(g_DoubleConstantsTest_constants.DOUBLE_LIST_TEST[i], EXPECTED_DOUBLE_LIST[i], EPSILON);
}
}
BOOST_AUTO_TEST_SUITE_END()

View File

@@ -178,7 +178,7 @@ private:
protected:
shared_ptr<server::TNonblockingServer> server;
private:
shared_ptr<concurrency::Thread> thread;
shared_ptr<apache::thrift::concurrency::Thread> thread;
};

View File

@@ -27,7 +27,7 @@
// The test weight, where 10 is 10 times more threads than baseline
// and the baseline is optimized for running in valgrind
static size_t WEIGHT = 10;
static int WEIGHT = 10;
int main(int argc, char** argv) {

View File

@@ -79,14 +79,13 @@ public:
= shared_ptr<TimerManagerTests::Task>(new TimerManagerTests::Task(_monitor, 10 * timeout));
{
TimerManager timerManager;
timerManager.threadFactory(shared_ptr<PlatformThreadFactory>(new PlatformThreadFactory()));
timerManager.start();
assert(timerManager.state() == TimerManager::STARTED);
if (timerManager.state() != TimerManager::STARTED) {
std::cerr << "timerManager is not in the STARTED state, but should be" << std::endl;
return false;
}
// Don't create task yet, because its constructor sets the expected completion time, and we
// need to delay between inserting the two tasks into the run queue.
@@ -94,34 +93,27 @@ public:
{
Synchronized s(_monitor);
timerManager.add(orphanTask, 10 * timeout);
try {
// Wait for 1 second in order to give timerManager a chance to start sleeping in response
// to adding orphanTask. We need to do this so we can verify that adding the second task
// kicks the dispatcher out of the current wait and starts the new 1 second wait.
_monitor.wait(1000);
assert(
0 == "ERROR: This wait should time out. TimerManager dispatcher may have a problem.");
} catch (TimedOutException&) {
}
THRIFT_SLEEP_USEC(timeout * 1000);
task.reset(new TimerManagerTests::Task(_monitor, timeout));
timerManager.add(task, timeout);
_monitor.wait();
}
assert(task->_done);
if (!task->_done) {
std::cerr << "task is not done, but it should have executed" << std::endl;
return false;
}
std::cout << "\t\t\t" << (task->_success ? "Success" : "Failure") << "!" << std::endl;
}
// timerManager.stop(); This is where it happens via destructor
assert(!orphanTask->_done);
if (orphanTask->_done) {
std::cerr << "orphan task is done, but it should not have executed" << std::endl;
return false;
}
return true;
}

View File

@@ -45,7 +45,7 @@
<UpdateRequired>false</UpdateRequired>
<MapFileExtensions>true</MapFileExtensions>
<ApplicationRevision>0</ApplicationRevision>
<ApplicationVersion>0.11.0.%2a</ApplicationVersion>
<ApplicationVersion>1.0.0.%2a</ApplicationVersion>
<IsWebBootstrapper>false</IsWebBootstrapper>
<UseApplicationTrust>false</UseApplicationTrust>
<BootstrapperEnabled>true</BootstrapperEnabled>

View File

@@ -51,5 +51,5 @@ using System.Runtime.InteropServices;
//
// You can specify all the values or you can default the Build and Revision Numbers
// by using the '*' as shown below:
[assembly: AssemblyVersion("0.11.0.1")]
[assembly: AssemblyFileVersion("0.11.0.1")]
[assembly: AssemblyVersion("1.0.0.1")]
[assembly: AssemblyFileVersion("1.0.0.1")]

View File

@@ -544,11 +544,6 @@ namespace Thrift.Protocol
int len = b.Length;
int off = 0;
// Ignore padding
int bound = len >= 2 ? len - 2 : 0;
for (int i = len - 1; i >= bound && b[i] == '='; --i) {
--len;
}
while (len >= 3)
{
// Encode 3 bytes at a time

View File

@@ -212,60 +212,78 @@ namespace Thrift.Server
/// <param name="threadContext"></param>
private void Execute(Object threadContext)
{
TTransport client = (TTransport)threadContext;
TProcessor processor = processorFactory.GetProcessor(client, this);
TTransport inputTransport = null;
TTransport outputTransport = null;
TProtocol inputProtocol = null;
TProtocol outputProtocol = null;
Object connectionContext = null;
try
using( TTransport client = (TTransport)threadContext)
{
inputTransport = inputTransportFactory.GetTransport(client);
outputTransport = outputTransportFactory.GetTransport(client);
inputProtocol = inputProtocolFactory.GetProtocol(inputTransport);
outputProtocol = outputProtocolFactory.GetProtocol(outputTransport);
//Recover event handler (if any) and fire createContext server event when a client connects
if (serverEventHandler != null)
connectionContext = serverEventHandler.createContext(inputProtocol, outputProtocol);
//Process client requests until client disconnects
while (!stop)
TProcessor processor = processorFactory.GetProcessor(client, this);
TTransport inputTransport = null;
TTransport outputTransport = null;
TProtocol inputProtocol = null;
TProtocol outputProtocol = null;
Object connectionContext = null;
try
{
if (!inputTransport.Peek())
break;
//Fire processContext server event
//N.B. This is the pattern implemented in C++ and the event fires provisionally.
//That is to say it may be many minutes between the event firing and the client request
//actually arriving or the client may hang up without ever makeing a request.
try
{
inputTransport = inputTransportFactory.GetTransport(client);
outputTransport = outputTransportFactory.GetTransport(client);
inputProtocol = inputProtocolFactory.GetProtocol(inputTransport);
outputProtocol = outputProtocolFactory.GetProtocol(outputTransport);
//Recover event handler (if any) and fire createContext server event when a client connects
if (serverEventHandler != null)
connectionContext = serverEventHandler.createContext(inputProtocol, outputProtocol);
//Process client requests until client disconnects
while (!stop)
{
if (!inputTransport.Peek())
break;
//Fire processContext server event
//N.B. This is the pattern implemented in C++ and the event fires provisionally.
//That is to say it may be many minutes between the event firing and the client request
//actually arriving or the client may hang up without ever makeing a request.
if (serverEventHandler != null)
serverEventHandler.processContext(connectionContext, inputTransport);
//Process client request (blocks until transport is readable)
if (!processor.Process(inputProtocol, outputProtocol))
break;
}
}
catch (TTransportException)
{
//Usually a client disconnect, expected
}
catch (Exception x)
{
//Unexpected
logDelegate("Error: " + x);
}
//Fire deleteContext server event after client disconnects
if (serverEventHandler != null)
serverEventHandler.processContext(connectionContext, inputTransport);
//Process client request (blocks until transport is readable)
if (!processor.Process(inputProtocol, outputProtocol))
break;
serverEventHandler.deleteContext(connectionContext, inputProtocol, outputProtocol);
}
finally
{
//Close transports
if (inputTransport != null)
inputTransport.Close();
if (outputTransport != null)
outputTransport.Close();
// disposable stuff should be disposed
if( inputProtocol != null)
inputProtocol.Dispose();
if( outputProtocol != null)
outputProtocol.Dispose();
if( inputTransport != null)
inputTransport.Dispose();
if( outputTransport != null)
outputTransport.Dispose();
}
}
catch (TTransportException)
{
//Usually a client disconnect, expected
}
catch (Exception x)
{
//Unexpected
logDelegate("Error: " + x);
}
//Fire deleteContext server event after client disconnects
if (serverEventHandler != null)
serverEventHandler.deleteContext(connectionContext, inputProtocol, outputProtocol);
//Close transports
if (inputTransport != null)
inputTransport.Close();
if (outputTransport != null)
outputTransport.Close();
}
public override void Stop()

View File

@@ -192,32 +192,33 @@ namespace Thrift.Server
private void ClientWorker(Object context)
{
TTransport client = (TTransport)context;
TProcessor processor = processorFactory.GetProcessor(client);
TTransport inputTransport = null;
TTransport outputTransport = null;
TProtocol inputProtocol = null;
TProtocol outputProtocol = null;
Object connectionContext = null;
try
{
using (inputTransport = inputTransportFactory.GetTransport(client))
using( TTransport client = (TTransport)context)
{
TProcessor processor = processorFactory.GetProcessor(client);
TTransport inputTransport = null;
TTransport outputTransport = null;
TProtocol inputProtocol = null;
TProtocol outputProtocol = null;
Object connectionContext = null;
try
{
using (outputTransport = outputTransportFactory.GetTransport(client))
try
{
inputTransport = inputTransportFactory.GetTransport(client);
outputTransport = outputTransportFactory.GetTransport(client);
inputProtocol = inputProtocolFactory.GetProtocol(inputTransport);
outputProtocol = outputProtocolFactory.GetProtocol(outputTransport);
//Recover event handler (if any) and fire createContext server event when a client connects
if (serverEventHandler != null)
connectionContext = serverEventHandler.createContext(inputProtocol, outputProtocol);
//Process client requests until client disconnects
while (!stop)
{
if (!inputTransport.Peek())
break;
//Fire processContext server event
//N.B. This is the pattern implemented in C++ and the event fires provisionally.
//That is to say it may be many minutes between the event firing and the client request
@@ -229,28 +230,42 @@ namespace Thrift.Server
break;
}
}
catch (TTransportException)
{
//Usually a client disconnect, expected
}
catch (Exception x)
{
//Unexpected
logDelegate("Error: " + x);
}
//Fire deleteContext server event after client disconnects
if (serverEventHandler != null)
serverEventHandler.deleteContext(connectionContext, inputProtocol, outputProtocol);
lock (clientLock)
{
clientThreads.Remove(Thread.CurrentThread);
Monitor.Pulse(clientLock);
}
}
finally
{
//Close transports
if (inputTransport != null)
inputTransport.Close();
if (outputTransport != null)
outputTransport.Close();
// disposable stuff should be disposed
if (inputProtocol != null)
inputProtocol.Dispose();
if (outputProtocol != null)
outputProtocol.Dispose();
}
}
catch (TTransportException)
{
//Usually a client disconnect, expected
}
catch (Exception x)
{
//Unexpected
logDelegate("Error: " + x);
}
//Fire deleteContext server event after client disconnects
if (serverEventHandler != null)
serverEventHandler.deleteContext(connectionContext, inputProtocol, outputProtocol);
lock (clientLock)
{
clientThreads.Remove(Thread.CurrentThread);
Monitor.Pulse(clientLock);
}
return;
}
public override void Stop()

View File

@@ -137,5 +137,10 @@ namespace Thrift
InvalidProtocol,
UnsupportedClientType
}
public ExceptionType Type
{
get { return type; }
}
}
}

View File

@@ -45,7 +45,7 @@
<UpdateRequired>false</UpdateRequired>
<MapFileExtensions>true</MapFileExtensions>
<ApplicationRevision>0</ApplicationRevision>
<ApplicationVersion>0.11.0.%2a</ApplicationVersion>
<ApplicationVersion>1.0.0.%2a</ApplicationVersion>
<IsWebBootstrapper>false</IsWebBootstrapper>
<UseApplicationTrust>false</UseApplicationTrust>
<BootstrapperEnabled>true</BootstrapperEnabled>

View File

@@ -51,5 +51,5 @@ using System.Runtime.InteropServices;
// Sie können alle Werte angeben oder die standardmäßigen Build- und Revisionsnummern
// übernehmen, indem Sie "*" eingeben:
// [assembly: AssemblyVersion("1.0.*")]
[assembly: AssemblyVersion("0.11.0.0")]
[assembly: AssemblyFileVersion("0.11.0.0")]
[assembly: AssemblyVersion("1.0.0.0")]
[assembly: AssemblyFileVersion("1.0.0.0")]

View File

@@ -46,7 +46,7 @@
<UpdateRequired>false</UpdateRequired>
<MapFileExtensions>true</MapFileExtensions>
<ApplicationRevision>0</ApplicationRevision>
<ApplicationVersion>0.11.0.%2a</ApplicationVersion>
<ApplicationVersion>1.0.0.%2a</ApplicationVersion>
<UseApplicationTrust>false</UseApplicationTrust>
<BootstrapperEnabled>true</BootstrapperEnabled>
</PropertyGroup>

View File

@@ -51,5 +51,5 @@ using System.Runtime.InteropServices;
// You can specify all the values or you can default the Build and Revision Numbers
// by using the '*' as shown below:
// [assembly: AssemblyVersion("1.0.*")]
[assembly: AssemblyVersion("0.11.0.0")]
[assembly: AssemblyFileVersion("0.11.0.0")]
[assembly: AssemblyVersion("1.0.0.0")]
[assembly: AssemblyFileVersion("1.0.0.0")]

View File

@@ -46,7 +46,7 @@
<UpdateRequired>false</UpdateRequired>
<MapFileExtensions>true</MapFileExtensions>
<ApplicationRevision>0</ApplicationRevision>
<ApplicationVersion>0.11.0.%2a</ApplicationVersion>
<ApplicationVersion>1.0.0.%2a</ApplicationVersion>
<UseApplicationTrust>false</UseApplicationTrust>
<BootstrapperEnabled>true</BootstrapperEnabled>
</PropertyGroup>

View File

@@ -51,5 +51,5 @@ using System.Runtime.InteropServices;
// You can specify all the values or you can default the Build and Revision Numbers
// by using the '*' as shown below:
// [assembly: AssemblyVersion("1.0.*")]
[assembly: AssemblyVersion("0.11.0.0")]
[assembly: AssemblyFileVersion("0.11.0.0")]
[assembly: AssemblyVersion("1.0.0.0")]
[assembly: AssemblyFileVersion("1.0.0.0")]

View File

@@ -49,5 +49,5 @@ using System.Runtime.InteropServices;
//
// You can specify all the values or you can default the Build and Revision Numbers
// by using the '*' as shown below:
[assembly: AssemblyVersion("0.11.0.1")]
[assembly: AssemblyFileVersion("0.11.0.1")]
[assembly: AssemblyVersion("1.0.0.1")]
[assembly: AssemblyFileVersion("1.0.0.1")]

View File

@@ -50,7 +50,7 @@ class TCompoundOperationException : TException {
/// The Thrift version string, used for informative purposes.
// Note: This is currently hardcoded, but will likely be filled in by the build
// system in future versions.
enum VERSION = "0.11.0";
enum VERSION = "1.0.0 dev";
/**
* Functions used for logging inside Thrift.

View File

@@ -140,7 +140,9 @@ class TSimpleServer : TServer {
}
}
} catch (TTransportException ttx) {
logError("Client died: %s", ttx);
if (ttx.type() != TTransportException.Type.END_OF_FILE) {
logError("Client died unexpectedly: %s", ttx);
}
} catch (Exception e) {
logError("Uncaught exception: %s", e);
}

View File

@@ -268,7 +268,9 @@ protected:
}
}
} catch (TTransportException ttx) {
logError("Client died: %s", ttx);
if (ttx.type() != TTransportException.Type.END_OF_FILE) {
logError("Client died unexpectedly: %s", ttx);
}
} catch (Exception e) {
logError("Uncaught exception: %s", e);
}

View File

@@ -173,7 +173,9 @@ private class WorkerThread : Thread {
}
}
} catch (TTransportException ttx) {
logError("Client died: %s", ttx);
if (ttx.type() != TTransportException.Type.END_OF_FILE) {
logError("Client died unexpectedly: %s", ttx);
}
} catch (Exception e) {
logError("Uncaught exception: %s", e);
}

View File

@@ -16,8 +16,11 @@
* specific language governing permissions and limitations
* under the License.
*/
module thrift_test_server;
import core.stdc.errno : errno;
import core.stdc.signal : signal, sigfn_t, SIGINT, SIG_DFL, SIG_ERR;
import core.thread : dur, Thread;
import std.algorithm;
import std.exception : enforce;
@@ -40,6 +43,7 @@ import thrift.transport.buffered;
import thrift.transport.framed;
import thrift.transport.http;
import thrift.transport.ssl;
import thrift.util.cancellation;
import thrift.util.hashset;
import test_utils;
@@ -205,14 +209,44 @@ private:
bool trace_;
}
shared(bool) gShutdown = false;
nothrow @nogc extern(C) void handleSignal(int sig) {
gShutdown = true;
}
// Runs a thread that waits for shutdown to be
// signaled and then triggers cancellation,
// causing the server to stop. While we could
// use a signalfd for this purpose, we are instead
// opting for a busy waiting scheme for maximum
// portability since signalfd is a linux thing.
class ShutdownThread : Thread {
this(TCancellationOrigin cancellation) {
cancellation_ = cancellation;
super(&run);
}
private:
void run() {
while (!gShutdown) {
Thread.sleep(dur!("msecs")(25));
}
cancellation_.trigger();
}
TCancellationOrigin cancellation_;
}
void main(string[] args) {
ushort port = 9090;
ServerType serverType;
ProtocolType protocolType;
size_t numIOThreads = 1;
TransportType transportType;
bool ssl;
bool trace;
bool ssl = false;
bool trace = true;
size_t taskPoolSize = totalCPUs;
getopt(args, "port", &port, "protocol", &protocolType, "server-type",
@@ -279,8 +313,26 @@ void main(string[] args) {
auto server = createServer(serverType, numIOThreads, taskPoolSize,
processor, serverSocket, transportFactory, protocolFactory);
// Set up SIGINT signal handling
sigfn_t oldHandler = signal(SIGINT, &handleSignal);
enforce(oldHandler != SIG_ERR,
"Could not replace the SIGINT signal handler: errno {0}".format(errno()));
// Set up a server cancellation trigger
auto cancel = new TCancellationOrigin();
// Set up a listener for the shutdown condition - this will
// wake up when the signal occurs and trigger cancellation.
auto shutdown = new ShutdownThread(cancel);
shutdown.start();
// Serve from this thread; the signal will stop the server
// and control will return here
writefln("Starting %s/%s %s ThriftTest server %son port %s...", protocolType,
transportType, serverType, ssl ? "(using SSL) ": "", port);
server.serve();
server.serve(cancel);
shutdown.join();
signal(SIGINT, SIG_DFL);
writeln("done.");
}

View File

@@ -25,7 +25,14 @@ class TFramedTransport extends TBufferedTransport {
final TTransport _transport;
final Uint8List headerBytes = new Uint8List(headerByteCount);
final Uint8List _headerBytes = new Uint8List(headerByteCount);
int _receivedHeaderBytes = 0;
int _bodySize = 0;
Uint8List _body = null;
int _receivedBodyBytes = 0;
Completer<Uint8List> _frameCompleter = null;
TFramedTransport(TTransport transport) : _transport = transport {
if (transport == null) {
@@ -51,33 +58,112 @@ class TFramedTransport extends TBufferedTransport {
if (got > 0) return got;
}
_readFrame();
// IMPORTANT: by the time you've got here,
// an entire frame is available for reading
return super.read(buffer, offset, length);
}
void _readFrame() {
_transport.readAll(headerBytes, 0, headerByteCount);
int size = headerBytes.buffer.asByteData().getUint32(0);
if (size < 0) {
throw new TTransportError(
TTransportErrorType.UNKNOWN, "Read a negative frame size: $size");
if (_body == null) {
bool gotFullHeader = _readFrameHeader();
if (!gotFullHeader) {
return;
}
}
Uint8List buffer = new Uint8List(size);
_transport.readAll(buffer, 0, size);
_setReadBuffer(buffer);
_readFrameBody();
}
bool _readFrameHeader() {
var remainingHeaderBytes = headerByteCount - _receivedHeaderBytes;
int got = _transport.read(_headerBytes, _receivedHeaderBytes, remainingHeaderBytes);
if (got < 0) {
throw new TTransportError(
TTransportErrorType.UNKNOWN, "Socket closed during frame header read");
}
_receivedHeaderBytes += got;
if (_receivedHeaderBytes == headerByteCount) {
int size = _headerBytes.buffer.asByteData().getUint32(0);
_receivedHeaderBytes = 0;
if (size < 0) {
throw new TTransportError(
TTransportErrorType.UNKNOWN, "Read a negative frame size: $size");
}
_bodySize = size;
_body = new Uint8List(_bodySize);
_receivedBodyBytes = 0;
return true;
} else {
_registerForReadableBytes();
return false;
}
}
void _readFrameBody() {
var remainingBodyBytes = _bodySize - _receivedBodyBytes;
int got = _transport.read(_body, _receivedBodyBytes, remainingBodyBytes);
if (got < 0) {
throw new TTransportError(
TTransportErrorType.UNKNOWN, "Socket closed during frame body read");
}
_receivedBodyBytes += got;
if (_receivedBodyBytes == _bodySize) {
var body = _body;
_bodySize = 0;
_body = null;
_receivedBodyBytes = 0;
_setReadBuffer(body);
var completer = _frameCompleter;
_frameCompleter = null;
completer.complete(new Uint8List(0));
} else {
_registerForReadableBytes();
}
}
Future flush() {
Uint8List buffer = consumeWriteBuffer();
int length = buffer.length;
if (_frameCompleter == null) {
Uint8List buffer = consumeWriteBuffer();
int length = buffer.length;
headerBytes.buffer.asByteData().setUint32(0, length);
_transport.write(headerBytes, 0, headerByteCount);
_transport.write(buffer, 0, length);
_headerBytes.buffer.asByteData().setUint32(0, length);
_transport.write(_headerBytes, 0, headerByteCount);
_transport.write(buffer, 0, length);
return _transport.flush();
_frameCompleter = new Completer<Uint8List>();
_registerForReadableBytes();
}
return _frameCompleter.future;
}
void _registerForReadableBytes() {
_transport.flush().then((_) {
_readFrame();
}).catchError((e) {
var completer = _frameCompleter;
_receivedHeaderBytes = 0;
_bodySize = 0;
_body = null;
_receivedBodyBytes = 0;
_frameCompleter = null;
completer.completeError(e);
});
}
}

View File

@@ -79,7 +79,9 @@ class TClientSocketTransport extends TSocketTransport {
var completer = new Completer<Uint8List>.sync();
_completers.add(completer);
socket.send(bytes);
if (bytes.lengthInBytes > 0) {
socket.send(bytes);
}
return completer.future;
}

View File

@@ -16,7 +16,7 @@
# under the License.
name: thrift
version: 0.11.0
version: 1.0.0-dev
description: >
A Dart library for Apache Thrift
author: Apache Thrift Developers <dev@thrift.apache.org>
@@ -33,7 +33,7 @@ dependencies:
dev_dependencies:
# test
mockito: ^0.11.0
mockito: ^1.0.0
test: ^0.12.0
# dart_dev - https://github.com/Workiva/dart_dev

View File

@@ -0,0 +1,175 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
library thrift.test.transport.t_framed_transport_test;
import 'dart:async';
import 'dart:convert';
import 'dart:typed_data' show Uint8List;
import 'package:test/test.dart';
import 'package:thrift/thrift.dart';
void main() {
group('TFramedTransport partial reads', () {
final flushAwaitDuration = new Duration(seconds: 10);
FakeReadOnlySocket socket;
TSocketTransport socketTransport;
TFramedTransport transport;
var messageAvailable;
setUp(() {
socket = new FakeReadOnlySocket();
socketTransport = new TClientSocketTransport(socket);
transport = new TFramedTransport(socketTransport);
messageAvailable = false;
});
expectNoReadableBytes() {
var readBuffer = new Uint8List(128);
var readBytes = transport.read(readBuffer, 0, readBuffer.lengthInBytes);
expect(readBytes, 0);
expect(messageAvailable, false);
}
test('Test transport reads messages where header and body are sent separately', () async {
// buffer into which we'll read
var readBuffer = new Uint8List(10);
var readBytes;
// registers for readable bytes
var flushFuture = transport.flush().timeout(flushAwaitDuration);
flushFuture.then((_) {
messageAvailable = true;
});
// write header bytes
socket.messageController.add(new Uint8List.fromList([0x00, 0x00, 0x00, 0x06]));
// you shouldn't be able to get any bytes from the read,
// because the header has been consumed internally
expectNoReadableBytes();
// write first batch of body
socket.messageController.add(new Uint8List.fromList(UTF8.encode("He")));
// you shouldn't be able to get any bytes from the read,
// because the frame has been consumed internally
expectNoReadableBytes();
// write second batch of body
socket.messageController.add(new Uint8List.fromList(UTF8.encode("llo!")));
// have to wait for the flush to complete,
// because it's only then that the frame is available for reading
await flushFuture;
expect(messageAvailable, true);
// at this point the frame is complete, so we expect the read to complete
readBytes = transport.read(readBuffer, 0, readBuffer.lengthInBytes);
expect(readBytes, 6);
expect(readBuffer.sublist(0, 6), UTF8.encode("Hello!"));
});
test('Test transport reads messages where header is sent in pieces '
'and body is also sent in pieces', () async {
// buffer into which we'll read
var readBuffer = new Uint8List(10);
var readBytes;
// registers for readable bytes
var flushFuture = transport.flush().timeout(flushAwaitDuration);
flushFuture.then((_) {
messageAvailable = true;
});
// write first part of header bytes
socket.messageController.add(new Uint8List.fromList([0x00, 0x00]));
// you shouldn't be able to get any bytes from the read
expectNoReadableBytes();
// write second part of header bytes
socket.messageController.add(new Uint8List.fromList([0x00, 0x03]));
// you shouldn't be able to get any bytes from the read again
// because only the header was read, and there's no frame body
readBytes = expectNoReadableBytes();
// write first batch of body
socket.messageController.add(new Uint8List.fromList(UTF8.encode("H")));
// you shouldn't be able to get any bytes from the read,
// because the frame has been consumed internally
expectNoReadableBytes();
// write second batch of body
socket.messageController.add(new Uint8List.fromList(UTF8.encode("i!")));
// have to wait for the flush to complete,
// because it's only then that the frame is available for reading
await flushFuture;
expect(messageAvailable, true);
// at this point the frame is complete, so we expect the read to complete
readBytes = transport.read(readBuffer, 0, readBuffer.lengthInBytes);
expect(readBytes, 3);
expect(readBuffer.sublist(0, 3), UTF8.encode("Hi!"));
});
});
}
class FakeReadOnlySocket extends TSocket {
StreamController<Uint8List> messageController = new StreamController<Uint8List>(sync: true);
StreamController<Object> errorController = new StreamController<Object>();
StreamController<TSocketState> stateController = new StreamController<TSocketState>();
@override
Future close() {
// noop
}
@override
bool get isClosed => false;
@override
bool get isOpen => true;
@override
Stream<Object> get onError => errorController.stream;
@override
Stream<Uint8List> get onMessage => messageController.stream;
@override
Stream<TSocketState> get onState => stateController.stream;
@override
Future open() {
// noop
}
@override
void send(Uint8List data) {
// noop
}
}

View File

@@ -53,11 +53,11 @@ uses
type
IMultiplexedProcessor = interface( IProcessor)
['{810FF32D-22A2-4D58-B129-B0590703ECEC}']
['{807F9D19-6CF4-4789-840E-93E87A12EB63}']
// Register a service with this TMultiplexedProcessor. This allows us
// to broker requests to individual services by using the service name
// to select them at request time.
procedure RegisterProcessor( const serviceName : String; const processor : IProcessor);
procedure RegisterProcessor( const serviceName : String; const processor : IProcessor; const asDefault : Boolean = FALSE);
end;
@@ -76,6 +76,7 @@ type
private
FServiceProcessorMap : TDictionary<String, IProcessor>;
FDefaultProcessor : IProcessor;
procedure Error( const oprot : IProtocol; const msg : TThriftMessage;
extype : TApplicationExceptionSpecializedClass; const etxt : string);
@@ -87,7 +88,7 @@ type
// Register a service with this TMultiplexedProcessorImpl. This allows us
// to broker requests to individual services by using the service name
// to select them at request time.
procedure RegisterProcessor( const serviceName : String; const processor : IProcessor);
procedure RegisterProcessor( const serviceName : String; const processor : IProcessor; const asDefault : Boolean = FALSE);
{ This implementation of process performs the following steps:
- Read the beginning of the message.
@@ -135,9 +136,15 @@ begin
end;
procedure TMultiplexedProcessorImpl.RegisterProcessor( const serviceName : String; const processor : IProcessor);
procedure TMultiplexedProcessorImpl.RegisterProcessor( const serviceName : String; const processor : IProcessor; const asDefault : Boolean);
begin
FServiceProcessorMap.Add( serviceName, processor);
if asDefault then begin
if FDefaultProcessor = nil
then FDefaultProcessor := processor
else raise TApplicationExceptionInternalError.Create('Only one default service allowed');
end;
end;
@@ -184,28 +191,37 @@ begin
end;
// Extract the service name
// use FDefaultProcessor as fallback if there is no separator
idx := Pos( TMultiplexedProtocol.SEPARATOR, msg.Name);
if idx < 1 then begin
if idx > 0 then begin
// Create a new TMessage, something that can be consumed by any TProtocol
sService := Copy( msg.Name, 1, idx-1);
if not FServiceProcessorMap.TryGetValue( sService, processor)
then begin
Error( oprot, msg,
TApplicationExceptionInternalError,
Format(ERROR_UNKNOWN_SERVICE,[sService]));
Exit( FALSE);
end;
// Create a new TMessage, removing the service name
Inc( idx, Length(TMultiplexedProtocol.SEPARATOR));
Init( newMsg, Copy( msg.Name, idx, MAXINT), msg.Type_, msg.SeqID);
end
else if FDefaultProcessor <> nil then begin
processor := FDefaultProcessor;
newMsg := msg; // no need to change
end
else begin
Error( oprot, msg,
TApplicationExceptionInvalidProtocol,
Format(ERROR_INCOMPATIBLE_PROT,[msg.Name]));
Exit( FALSE);
end;
// Create a new TMessage, something that can be consumed by any TProtocol
sService := Copy( msg.Name, 1, idx-1);
if not FServiceProcessorMap.TryGetValue( sService, processor)
then begin
Error( oprot, msg,
TApplicationExceptionInternalError,
Format(ERROR_UNKNOWN_SERVICE,[sService]));
Exit( FALSE);
end;
// Create a new TMessage, removing the service name
Inc( idx, Length(TMultiplexedProtocol.SEPARATOR));
Init( newMsg, Copy( msg.Name, idx, MAXINT), msg.Type_, msg.SeqID);
// Dispatch processing to the stored processor
protocol := TStoredMessageProtocol.Create( iprot, newMsg);
result := processor.process( protocol, oprot, events);
@@ -213,4 +229,3 @@ end;
end.

View File

@@ -132,6 +132,7 @@ begin
end;
function TThriftStreamAdapterCOM.Read( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer;
var pTmp : PByte;
begin
inherited;
@@ -141,7 +142,9 @@ begin
Result := 0;
if FStream <> nil then begin
if count > 0 then begin
FStream.Read( @(PByteArray(pBuf)^[offset]), count, @Result);
pTmp := pBuf;
Inc( pTmp, offset);
FStream.Read( pTmp, count, @Result);
end;
end;
end;
@@ -172,11 +175,14 @@ end;
procedure TThriftStreamAdapterCOM.Write( const pBuf: Pointer; offset: Integer; count: Integer);
var nWritten : Integer;
pTmp : PByte;
begin
inherited;
if IsOpen then begin
if count > 0 then begin
FStream.Write( @(PByteArray(pBuf)^[offset]), count, @nWritten);
pTmp := pBuf;
Inc( pTmp, offset);
FStream.Write( pTmp, count, @nWritten);
end;
end;
end;
@@ -259,14 +265,18 @@ begin
end;
function TThriftStreamAdapterDelphi.Read(const pBuf : Pointer; const buflen : Integer; offset, count: Integer): Integer;
var pTmp : PByte;
begin
inherited;
if count >= buflen-offset
then count := buflen-offset;
if count > 0
then Result := FStream.Read( PByteArray(pBuf)^[offset], count)
if count > 0 then begin
pTmp := pBuf;
Inc( pTmp, offset);
Result := FStream.Read( pTmp^, count)
end
else Result := 0;
end;
@@ -296,10 +306,13 @@ begin
end;
procedure TThriftStreamAdapterDelphi.Write(const pBuf : Pointer; offset, count: Integer);
var pTmp : PByte;
begin
inherited;
if count > 0 then begin
FStream.Write( PByteArray(pBuf)^[offset], count)
pTmp := pBuf;
Inc( pTmp, offset);
FStream.Write( pTmp^, count)
end;
end;

View File

@@ -328,6 +328,7 @@ end;
procedure TPipeStreamBase.WriteDirect( const pBuf : Pointer; offset: Integer; count: Integer);
var cbWritten, nBytes : DWORD;
pData : PByte;
begin
if not IsOpen
then raise TTransportExceptionNotOpen.Create('Called write on non-open pipe');
@@ -336,11 +337,13 @@ begin
// there's a system limit around 0x10000 bytes that we hit otherwise
// MSDN: "Pipe write operations across a network are limited to 65,535 bytes per write. For more information regarding pipes, see the Remarks section."
nBytes := Min( 15*4096, count); // 16 would exceed the limit
pData := pBuf;
Inc( pData, offset);
while nBytes > 0 do begin
if not WriteFile( FPipe, PByteArray(pBuf)^[offset], nBytes, cbWritten, nil)
if not WriteFile( FPipe, pData^, nBytes, cbWritten, nil)
then raise TTransportExceptionNotOpen.Create('Write to pipe failed');
Inc( offset, cbWritten);
Inc( pData, cbWritten);
Dec( count, cbWritten);
nBytes := Min( nBytes, count);
end;
@@ -350,6 +353,7 @@ end;
procedure TPipeStreamBase.WriteOverlapped( const pBuf : Pointer; offset: Integer; count: Integer);
var cbWritten, dwWait, dwError, nBytes : DWORD;
overlapped : IOverlappedHelper;
pData : PByte;
begin
if not IsOpen
then raise TTransportExceptionNotOpen.Create('Called write on non-open pipe');
@@ -358,17 +362,21 @@ begin
// there's a system limit around 0x10000 bytes that we hit otherwise
// MSDN: "Pipe write operations across a network are limited to 65,535 bytes per write. For more information regarding pipes, see the Remarks section."
nBytes := Min( 15*4096, count); // 16 would exceed the limit
pData := pBuf;
Inc( pData, offset);
while nBytes > 0 do begin
overlapped := TOverlappedHelperImpl.Create;
if not WriteFile( FPipe, PByteArray(pBuf)^[offset], nBytes, cbWritten, overlapped.OverlappedPtr)
if not WriteFile( FPipe, pData^, nBytes, cbWritten, overlapped.OverlappedPtr)
then begin
dwError := GetLastError;
case dwError of
ERROR_IO_PENDING : begin
dwWait := overlapped.WaitFor(FTimeout);
if (dwWait = WAIT_TIMEOUT)
then raise TTransportExceptionTimedOut.Create('Pipe write timed out');
if (dwWait = WAIT_TIMEOUT) then begin
CancelIo( FPipe); // prevents possible AV on invalid overlapped ptr
raise TTransportExceptionTimedOut.Create('Pipe write timed out');
end;
if (dwWait <> WAIT_OBJECT_0)
or not GetOverlappedResult( FPipe, overlapped.Overlapped, cbWritten, TRUE)
@@ -382,7 +390,7 @@ begin
ASSERT( DWORD(nBytes) = cbWritten);
Inc( offset, cbWritten);
Inc( pData, cbWritten);
Dec( count, cbWritten);
nBytes := Min( nBytes, count);
end;
@@ -393,6 +401,7 @@ function TPipeStreamBase.ReadDirect( const pBuf : Pointer; const buflen : In
var cbRead, dwErr, nRemaining : DWORD;
bytes, retries : LongInt;
bOk : Boolean;
pData : PByte;
const INTERVAL = 10; // ms
begin
if not IsOpen
@@ -427,14 +436,16 @@ begin
result := 0;
nRemaining := count;
pData := pBuf;
Inc( pData, offset);
while nRemaining > 0 do begin
// read the data (or block INFINITE-ly)
bOk := ReadFile( FPipe, PByteArray(pBuf)^[offset], nRemaining, cbRead, nil);
bOk := ReadFile( FPipe, pData^, nRemaining, cbRead, nil);
if (not bOk) and (GetLastError() <> ERROR_MORE_DATA)
then Break; // No more data, possibly because client disconnected.
Dec( nRemaining, cbRead);
Inc( offset, cbRead);
Inc( pData, cbRead);
Inc( result, cbRead);
end;
end;
@@ -444,25 +455,30 @@ function TPipeStreamBase.ReadOverlapped( const pBuf : Pointer; const buflen : In
var cbRead, dwWait, dwError, nRemaining : DWORD;
bOk : Boolean;
overlapped : IOverlappedHelper;
pData : PByte;
begin
if not IsOpen
then raise TTransportExceptionNotOpen.Create('Called read on non-open pipe');
result := 0;
nRemaining := count;
pData := pBuf;
Inc( pData, offset);
while nRemaining > 0 do begin
overlapped := TOverlappedHelperImpl.Create;
// read the data
bOk := ReadFile( FPipe, PByteArray(pBuf)^[offset], nRemaining, cbRead, overlapped.OverlappedPtr);
bOk := ReadFile( FPipe, pData^, nRemaining, cbRead, overlapped.OverlappedPtr);
if not bOk then begin
dwError := GetLastError;
case dwError of
ERROR_IO_PENDING : begin
dwWait := overlapped.WaitFor(FTimeout);
if (dwWait = WAIT_TIMEOUT)
then raise TTransportExceptionTimedOut.Create('Pipe read timed out');
if (dwWait = WAIT_TIMEOUT) then begin
CancelIo( FPipe); // prevents possible AV on invalid overlapped ptr
raise TTransportExceptionTimedOut.Create('Pipe read timed out');
end;
if (dwWait <> WAIT_OBJECT_0)
or not GetOverlappedResult( FPipe, overlapped.Overlapped, cbRead, TRUE)
@@ -477,7 +493,7 @@ begin
ASSERT( cbRead > 0); // see TTransportImpl.ReadAll()
ASSERT( cbRead <= DWORD(nRemaining));
Dec( nRemaining, cbRead);
Inc( offset, cbRead);
Inc( pData, cbRead);
Inc( result, cbRead);
end;
end;
@@ -864,8 +880,10 @@ begin
CreateNamedPipe;
while not FConnected do begin
if QueryStopServer
then Abort;
if QueryStopServer then begin
InternalClose;
Abort;
end;
if Assigned(fnAccepting)
then fnAccepting();

View File

@@ -966,10 +966,11 @@ function TBufferedStreamImpl.Read( const pBuf : Pointer; const buflen : Integer;
var
nRead : Integer;
tempbuf : TBytes;
pTmp : PByte;
begin
inherited;
Result := 0;
if IsOpen then begin
while count > 0 do begin
@@ -984,8 +985,10 @@ begin
end;
if FReadBuffer.Position < FReadBuffer.Size then begin
nRead := Min( FReadBuffer.Size - FReadBuffer.Position, count);
Inc( Result, FReadBuffer.Read( PByteArray(pBuf)^[offset], nRead));
nRead := Min( FReadBuffer.Size - FReadBuffer.Position, count);
pTmp := pBuf;
Inc( pTmp, offset);
Inc( Result, FReadBuffer.Read( pTmp^, nRead));
Dec( count, nRead);
Inc( offset, nRead);
end;
@@ -1011,11 +1014,14 @@ begin
end;
procedure TBufferedStreamImpl.Write( const pBuf : Pointer; offset: Integer; count: Integer);
var pTmp : PByte;
begin
inherited;
if count > 0 then begin
if IsOpen then begin
FWriteBuffer.Write( PByteArray(pBuf)^[offset], count );
pTmp := pBuf;
Inc( pTmp, offset);
FWriteBuffer.Write( pTmp^, count );
if FWriteBuffer.Size > FBufSize then begin
Flush;
end;
@@ -1066,7 +1072,7 @@ end;
function TStreamTransportImpl.GetOutputStream: IThriftStream;
begin
Result := FInputStream;
Result := FOutputStream;
end;
procedure TStreamTransportImpl.Open;
@@ -1254,12 +1260,16 @@ begin
end;
function TFramedTransportImpl.Read( const pBuf : Pointer; const buflen : Integer; off: Integer; len: Integer): Integer;
var pTmp : PByte;
begin
if len > (buflen-off)
then len := buflen-off;
pTmp := pBuf;
Inc( pTmp, off);
if (FReadBuffer <> nil) and (len > 0) then begin
result := FReadBuffer.Read( PByteArray(pBuf)^[off], len);
result := FReadBuffer.Read( pTmp^, len);
if result > 0 then begin
Exit;
end;
@@ -1267,7 +1277,7 @@ begin
ReadFrame;
if len > 0
then Result := FReadBuffer.Read( PByteArray(pBuf)^[off], len)
then Result := FReadBuffer.Read( pTmp^, len)
else Result := 0;
end;
@@ -1294,9 +1304,14 @@ begin
end;
procedure TFramedTransportImpl.Write( const pBuf : Pointer; off, len : Integer);
var pTmp : PByte;
begin
if len > 0
then FWriteBuffer.Write( PByteArray(pBuf)^[off], len );
if len > 0 then begin
pTmp := pBuf;
Inc( pTmp, off);
FWriteBuffer.Write( pTmp^, len );
end;
end;
{ TFramedTransport.TFactory }
@@ -1482,7 +1497,7 @@ var wfd : TWaitForData;
wsaError,
msecs : Integer;
nBytes : Integer;
pDest : PByte;
pTmp : PByte;
begin
inherited;
@@ -1491,11 +1506,12 @@ begin
else msecs := DEFAULT_THRIFT_TIMEOUT;
result := 0;
pDest := @(PByteArray(pBuf)^[offset]);
pTmp := pBuf;
Inc( pTmp, offset);
while count > 0 do begin
while TRUE do begin
wfd := WaitForData( msecs, pDest, count, wsaError, nBytes);
wfd := WaitForData( msecs, pTmp, count, wsaError, nBytes);
case wfd of
TWaitForData.wfd_Error : Exit;
TWaitForData.wfd_HaveData : Break;
@@ -1519,8 +1535,8 @@ begin
msecs := Max( msecs, 200);
ASSERT( nBytes <= count);
nBytes := FTcpClient.ReceiveBuf( pDest^, nBytes);
Inc( pDest, nBytes);
nBytes := FTcpClient.ReceiveBuf( pTmp^, nBytes);
Inc( pTmp, nBytes);
Dec( count, nBytes);
Inc( result, nBytes);
end;
@@ -1546,6 +1562,7 @@ procedure TTcpSocketStreamImpl.Write( const pBuf : Pointer; offset, count: Integ
// old sockets version
var bCanWrite, bError : Boolean;
retval, wsaError : Integer;
pTmp : PByte;
begin
inherited;
@@ -1566,7 +1583,9 @@ begin
if bError or not bCanWrite
then raise TTransportExceptionUnknown.Create('unknown error');
FTcpClient.SendBuf( PByteArray(pBuf)^[offset], count);
pTmp := pBuf;
Inc( pTmp, offset);
FTcpClient.SendBuf( pTmp^, count);
end;
{$ELSE}
@@ -1574,16 +1593,17 @@ end;
function TTcpSocketStreamImpl.Read( const pBuf : Pointer; const buflen : Integer; offset: Integer; count: Integer): Integer;
// new sockets version
var nBytes : Integer;
pDest : PByte;
pTmp : PByte;
begin
inherited;
result := 0;
pDest := @(PByteArray(pBuf)^[offset]);
pTmp := pBuf;
Inc( pTmp, offset);
while count > 0 do begin
nBytes := FTcpClient.Read(pDest^, count);
nBytes := FTcpClient.Read( pTmp^, count);
if nBytes = 0 then Exit;
Inc( pDest, nBytes);
Inc( pTmp, nBytes);
Dec( count, nBytes);
Inc( result, nBytes);
end;
@@ -1610,13 +1630,16 @@ end;
procedure TTcpSocketStreamImpl.Write( const pBuf : Pointer; offset, count: Integer);
// new sockets version
var pTmp : PByte;
begin
inherited;
if not FTcpClient.IsOpen
then raise TTransportExceptionNotOpen.Create('not open');
FTcpClient.Write( PByteArray(pBuf)^[offset], count);
pTmp := pBuf;
Inc( pTmp, offset);
FTcpClient.Write( pTmp^, count);
end;
{$ENDIF}

View File

@@ -206,8 +206,12 @@ end;
class function CharUtils.IsHighSurrogate( const c : Char) : Boolean;
begin
{$IF CompilerVersion < 23.0}
result := Character.IsHighSurrogate( c);
{$IF CompilerVersion < 25.0}
{$IFDEF OLD_UNIT_NAMES}
result := Character.IsHighSurrogate(c);
{$ELSE}
result := System.Character.IsHighSurrogate(c);
{$ENDIF}
{$ELSE}
result := c.IsHighSurrogate();
{$IFEND}
@@ -216,10 +220,14 @@ end;
class function CharUtils.IsLowSurrogate( const c : Char) : Boolean;
begin
{$IF CompilerVersion < 23.0}
result := Character.IsLowSurrogate( c);
{$IF CompilerVersion < 25.0}
{$IFDEF OLD_UNIT_NAMES}
result := Character.IsLowSurrogate(c);
{$ELSE}
result := System.Character.IsLowSurrogate(c);
{$ENDIF}
{$ELSE}
result := c.IsLowSurrogate;
result := c.IsLowSurrogate();
{$IFEND}
end;

View File

@@ -25,7 +25,7 @@ uses
SysUtils, Thrift.Protocol;
const
Version = '0.11.0';
Version = '1.0.0-dev';
type
TApplicationExceptionSpecializedClass = class of TApplicationExceptionSpecialized;

View File

@@ -17,7 +17,7 @@
* under the License.
*)
unit Thrift.Console;
unit ConsoleHelper;
interface
@@ -99,10 +99,9 @@ begin
begin
idx := FMemo.Count - 1;
if idx < 0 then
begin
FMemo.Add( S );
end;
FMemo[idx] := FMemo[idx] + S;
FMemo.Add( S )
else
FMemo[idx] := FMemo[idx] + S;
end;
FLineBreak := bWriteLine;
end;

View File

@@ -25,13 +25,19 @@ unit TestClient;
{.$DEFINE PerfTest} // activate the performance test
{$DEFINE Exceptions} // activate the exceptions test (or disable while debugging)
{$if CompilerVersion >= 28}
{$DEFINE SupportsAsync}
{$ifend}
interface
uses
Windows, SysUtils, Classes, Math,
{$IFDEF SupportsAsync} System.Threading, {$ENDIF}
DateUtils,
Generics.Collections,
TestConstants,
ConsoleHelper,
Thrift,
Thrift.Protocol.Compact,
Thrift.Protocol.JSON,
@@ -41,8 +47,7 @@ uses
Thrift.Stream,
Thrift.Test,
Thrift.Utils,
Thrift.Collections,
Thrift.Console;
Thrift.Collections;
type
TThreadConsole = class
@@ -66,6 +71,13 @@ type
);
TTestGroups = set of TTestGroup;
TTestSize = (
Empty, // Edge case: the zero-length empty binary
Normal, // Fairly small array of usual size (256 bytes)
ByteArrayTest, // THRIFT-4454 Large writes/reads may cause range check errors in debug mode
PipeWriteLimit // THRIFT-4372 Pipe write operations across a network are limited to 65,535 bytes per write.
);
private
FTransport : ITransport;
FProtocol : IProtocol;
@@ -85,8 +97,12 @@ type
function CalculateExitCode : Byte;
procedure ClientTest;
{$IFDEF SupportsAsync}
procedure ClientAsyncTest;
{$ENDIF}
procedure JSONProtocolReadWriteTest;
function PrepareBinaryData( aRandomDist, aHuge : Boolean) : TBytes;
function PrepareBinaryData( aRandomDist : Boolean; aSize : TTestSize) : TBytes;
{$IFDEF StressTest}
procedure StressTest(const client : TThriftTest.Iface);
{$ENDIF}
@@ -177,6 +193,7 @@ end;
class function TTestClient.Execute(const args: array of string) : Byte;
var
i : Integer;
threadExitCode : Byte;
host : string;
port : Integer;
sPipeName : string;
@@ -374,11 +391,13 @@ begin
result := 0;
for test := 0 to FNumThread - 1 do begin
result := result or threads[test].WaitFor;
threadExitCode := threads[test].WaitFor;
result := result or threadExitCode;
end;
for test := 0 to FNumThread - 1
do threads[test].Free;
for test := 0 to FNumThread - 1 do begin
threads[test].Free;
end;
Console.Write('Total time: ' + IntToStr( MilliSecondsBetween(Now, dtStart)));
@@ -455,6 +474,7 @@ var
first_map : IThriftDictionary<TNumberz, IInsanity>;
second_map : IThriftDictionary<TNumberz, IInsanity>;
pair : TPair<TNumberz, TUserId>;
testsize : TTestSize;
begin
client := TThriftTest.TClient.Create( FProtocol);
FTransport.Open;
@@ -547,42 +567,18 @@ begin
Expect( i64 = -34359738368, 'testI64(-34359738368) = ' + IntToStr( i64));
// random binary small
binOut := PrepareBinaryData( TRUE, FALSE);
Console.WriteLine('testBinary('+BytesToHex(binOut)+')');
try
binIn := client.testBinary(binOut);
Expect( Length(binOut) = Length(binIn), 'testBinary(): length '+IntToStr(Length(binOut))+' = '+IntToStr(Length(binIn)));
i32 := Min( Length(binOut), Length(binIn));
Expect( CompareMem( binOut, binIn, i32), 'testBinary('+BytesToHex(binOut)+') = '+BytesToHex(binIn));
except
on e:TApplicationException do Console.WriteLine('testBinary(): '+e.Message);
on e:Exception do Expect( FALSE, 'testBinary(): Unexpected exception "'+e.ClassName+'": '+e.Message);
end;
// random binary huge
binOut := PrepareBinaryData( TRUE, TRUE);
Console.WriteLine('testBinary('+BytesToHex(binOut)+')');
try
binIn := client.testBinary(binOut);
Expect( Length(binOut) = Length(binIn), 'testBinary(): length '+IntToStr(Length(binOut))+' = '+IntToStr(Length(binIn)));
i32 := Min( Length(binOut), Length(binIn));
Expect( CompareMem( binOut, binIn, i32), 'testBinary('+BytesToHex(binOut)+') = '+BytesToHex(binIn));
except
on e:TApplicationException do Console.WriteLine('testBinary(): '+e.Message);
on e:Exception do Expect( FALSE, 'testBinary(): Unexpected exception "'+e.ClassName+'": '+e.Message);
end;
// empty binary
SetLength( binOut, 0);
Console.WriteLine('testBinary('+BytesToHex(binOut)+')');
try
binIn := client.testBinary(binOut);
Expect( Length(binOut) = Length(binIn), 'testBinary(): length '+IntToStr(Length(binOut))+' = '+IntToStr(Length(binIn)));
i32 := Min( Length(binOut), Length(binIn));
Expect( CompareMem( binOut, binIn, i32), 'testBinary('+BytesToHex(binOut)+') = '+BytesToHex(binIn));
except
on e:TApplicationException do Console.WriteLine('testBinary(): '+e.Message);
on e:Exception do Expect( FALSE, 'testBinary(): Unexpected exception "'+e.ClassName+'": '+e.Message);
for testsize := Low(TTestSize) to High(TTestSize) do begin
binOut := PrepareBinaryData( TRUE, testsize);
Console.WriteLine('testBinary('+BytesToHex(binOut)+')');
try
binIn := client.testBinary(binOut);
Expect( Length(binOut) = Length(binIn), 'testBinary(): length '+IntToStr(Length(binOut))+' = '+IntToStr(Length(binIn)));
i32 := Min( Length(binOut), Length(binIn));
Expect( CompareMem( binOut, binIn, i32), 'testBinary('+BytesToHex(binOut)+') = '+BytesToHex(binIn));
except
on e:TApplicationException do Console.WriteLine('testBinary(): '+e.Message);
on e:Exception do Expect( FALSE, 'testBinary(): Unexpected exception "'+e.ClassName+'": '+e.Message);
end;
end;
Console.WriteLine('testDouble(5.325098235)');
@@ -1004,6 +1000,33 @@ begin
end;
{$IFDEF SupportsAsync}
procedure TClientThread.ClientAsyncTest;
var
client : TThriftTest.IAsync;
s : string;
i8 : ShortInt;
begin
StartTestGroup( 'Async Tests', test_Unknown);
client := TThriftTest.TClient.Create( FProtocol);
FTransport.Open;
// oneway void functions
client.testOnewayAsync(1).Wait;
Expect( TRUE, 'Test Oneway(1)'); // success := no exception
// normal functions
s := client.testStringAsync(HUGE_TEST_STRING).Value;
Expect( length(s) = length(HUGE_TEST_STRING),
'testString( length(HUGE_TEST_STRING) = '+IntToStr(Length(HUGE_TEST_STRING))+') '
+'=> length(result) = '+IntToStr(Length(s)));
i8 := client.testByte(1).Value;
Expect( i8 = 1, 'testByte(1) = ' + IntToStr( i8 ));
end;
{$ENDIF}
{$IFDEF StressTest}
procedure TClientThread.StressTest(const client : TThriftTest.Iface);
begin
@@ -1024,18 +1047,25 @@ end;
{$ENDIF}
function TClientThread.PrepareBinaryData( aRandomDist, aHuge : Boolean) : TBytes;
function TClientThread.PrepareBinaryData( aRandomDist : Boolean; aSize : TTestSize) : TBytes;
var i : Integer;
begin
if aHuge
then SetLength( result, $12345) // tests for THRIFT-4372
else SetLength( result, $100);
case aSize of
Empty : SetLength( result, 0);
Normal : SetLength( result, $100);
ByteArrayTest : SetLength( result, SizeOf(TByteArray) + 128);
PipeWriteLimit : SetLength( result, 65535 + 128);
else
raise EArgumentException.Create('aSize');
end;
ASSERT( Low(result) = 0);
if Length(result) = 0 then Exit;
// linear distribution, unless random is requested
if not aRandomDist then begin
for i := Low(result) to High(result) do begin
result[i] := i;
result[i] := i mod $100;
end;
Exit;
end;
@@ -1090,7 +1120,7 @@ begin
StartTestGroup( 'JsonProtocolTest', test_Unknown);
// prepare binary data
binary := PrepareBinaryData( FALSE, FALSE);
binary := PrepareBinaryData( FALSE, Normal);
SetLength( emptyBinary, 0); // empty binary data block
// output setup
@@ -1303,12 +1333,15 @@ begin
try
{$IFDEF Win64}
UseInterlockedExchangeAdd64;
{$ENDIF}
{$ENDIF}
JSONProtocolReadWriteTest;
for i := 0 to FNumIteration - 1 do
begin
ClientTest;
{$IFDEF SupportsAsync}
ClientAsyncTest;
{$ENDIF}
end;
except
on e:Exception do Expect( FALSE, 'unexpected exception: "'+e.message+'"');

View File

@@ -29,7 +29,6 @@ interface
uses
Windows, SysUtils,
Generics.Collections,
Thrift.Console,
Thrift.Server,
Thrift.Transport,
Thrift.Transport.Pipes,
@@ -42,6 +41,7 @@ uses
Thrift,
TestConstants,
TestServerEvents,
ConsoleHelper,
Contnrs;
type

View File

@@ -27,7 +27,7 @@ uses
Thrift.Protocol,
Thrift.Transport,
Thrift.Server,
Thrift.Console;
ConsoleHelper;
type
TRequestEventsImpl = class( TInterfacedObject, IRequestEvents)

View File

@@ -37,7 +37,6 @@ uses
Thrift.Collections in '..\src\Thrift.Collections.pas',
Thrift.Server in '..\src\Thrift.Server.pas',
Thrift.Stream in '..\src\Thrift.Stream.pas',
Thrift.Console in '..\src\Thrift.Console.pas',
Thrift.TypeRegistry in '..\src\Thrift.TypeRegistry.pas',
Thrift.Utils in '..\src\Thrift.Utils.pas';

View File

@@ -28,7 +28,6 @@ interface
uses
Windows, SysUtils,
Generics.Collections,
Thrift.Console,
Thrift.Server,
Thrift.Transport,
Thrift.Transport.Pipes,
@@ -41,6 +40,7 @@ uses
Benchmark, // in gen-delphi folder
Aggr, // in gen-delphi folder
Multiplex.Test.Common,
ConsoleHelper,
Contnrs;
type

View File

@@ -34,7 +34,6 @@ uses
Thrift.Collections in '..\..\src\Thrift.Collections.pas',
Thrift.Server in '..\..\src\Thrift.Server.pas',
Thrift.Stream in '..\..\src\Thrift.Stream.pas',
Thrift.Console in '..\..\src\Thrift.Console.pas',
Thrift.TypeRegistry in '..\..\src\Thrift.TypeRegistry.pas',
Thrift.Utils in '..\..\src\Thrift.Utils.pas';

View File

@@ -24,6 +24,7 @@ program Multiplex.Test.Server;
uses
SysUtils,
Multiplex.Server.Main in 'Multiplex.Server.Main.pas',
ConsoleHelper in '..\ConsoleHelper.pas',
Thrift in '..\..\src\Thrift.pas',
Thrift.Socket in '..\..\src\Thrift.Socket.pas',
Thrift.Transport in '..\..\src\Thrift.Transport.pas',
@@ -33,7 +34,6 @@ uses
Thrift.Processor.Multiplex in '..\..\src\Thrift.Processor.Multiplex.pas',
Thrift.Collections in '..\..\src\Thrift.Collections.pas',
Thrift.Server in '..\..\src\Thrift.Server.pas',
Thrift.Console in '..\..\src\Thrift.Console.pas',
Thrift.Utils in '..\..\src\Thrift.Utils.pas',
Thrift.TypeRegistry in '..\..\src\Thrift.TypeRegistry.pas',
Thrift.Stream in '..\..\src\Thrift.Stream.pas';

View File

@@ -30,7 +30,6 @@ uses
Thrift.Protocol.JSON in '..\..\src\Thrift.Protocol.JSON.pas',
Thrift.Collections in '..\..\src\Thrift.Collections.pas',
Thrift.Server in '..\..\src\Thrift.Server.pas',
Thrift.Console in '..\..\src\Thrift.Console.pas',
Thrift.Utils in '..\..\src\Thrift.Utils.pas',
Thrift.Serializer in '..\..\src\Thrift.Serializer.pas',
Thrift.Stream in '..\..\src\Thrift.Stream.pas',

View File

@@ -37,7 +37,6 @@ uses
Thrift.Processor.Multiplex in '..\src\Thrift.Processor.Multiplex.pas',
Thrift.Collections in '..\src\Thrift.Collections.pas',
Thrift.Server in '..\src\Thrift.Server.pas',
Thrift.Console in '..\src\Thrift.Console.pas',
Thrift.TypeRegistry in '..\src\Thrift.TypeRegistry.pas',
Thrift.Utils in '..\src\Thrift.Utils.pas',
Thrift.Stream in '..\src\Thrift.Stream.pas';

View File

@@ -31,7 +31,6 @@ uses
Thrift.Protocol.JSON in '..\..\src\Thrift.Protocol.JSON.pas',
Thrift.Collections in '..\..\src\Thrift.Collections.pas',
Thrift.Server in '..\..\src\Thrift.Server.pas',
Thrift.Console in '..\..\src\Thrift.Console.pas',
Thrift.Utils in '..\..\src\Thrift.Utils.pas',
Thrift.TypeRegistry in '..\..\src\Thrift.TypeRegistry.pas',
Thrift.Stream in '..\..\src\Thrift.Stream.pas';

View File

@@ -31,7 +31,6 @@ uses
Thrift.Protocol.JSON in '..\..\src\Thrift.Protocol.JSON.pas',
Thrift.Collections in '..\..\src\Thrift.Collections.pas',
Thrift.Server in '..\..\src\Thrift.Server.pas',
Thrift.Console in '..\..\src\Thrift.Console.pas',
Thrift.Utils in '..\..\src\Thrift.Utils.pas',
Thrift.TypeRegistry in '..\..\src\Thrift.TypeRegistry.pas',
Thrift.Stream in '..\..\src\Thrift.Stream.pas';

View File

@@ -30,7 +30,6 @@ uses
Thrift.Protocol.JSON in '..\..\src\Thrift.Protocol.JSON.pas',
Thrift.Collections in '..\..\src\Thrift.Collections.pas',
Thrift.Server in '..\..\src\Thrift.Server.pas',
Thrift.Console in '..\..\src\Thrift.Console.pas',
Thrift.Utils in '..\..\src\Thrift.Utils.pas',
Thrift.Serializer in '..\..\src\Thrift.Serializer.pas',
Thrift.Stream in '..\..\src\Thrift.Stream.pas',

View File

@@ -23,6 +23,7 @@ THRIFT_FILES = $(wildcard test/*.thrift) \
$(THRIFT_OMIT_FILE) \
../../test/ConstantsDemo.thrift \
../../test/NameConflictTest.thrift \
../../test/DoubleConstantsTest.thrift \
../../test/ThriftTest.thrift
if ERLANG_OTP16

View File

@@ -0,0 +1,68 @@
%%
%% Licensed to the Apache Software Foundation (ASF) under one
%% or more contributor license agreements. See the NOTICE file
%% distributed with this work for additional information
%% regarding copyright ownership. The ASF licenses this file
%% to you under the Apache License, Version 2.0 (the
%% "License"); you may not use this file except in compliance
%% with the License. You may obtain a copy of the License at
%%
%% http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing,
%% software distributed under the License is distributed on an
%% "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
%% KIND, either express or implied. See the License for the
%% specific language governing permissions and limitations
%% under the License.
%%
-module(test_rendered_double_constants).
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-include("gen-erl/double_constants_test_constants.hrl").
-define(EPSILON, 0.0000001).
rendered_double_constants_test() ->
?assert(abs(1.0 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_INT_CONSTANT_TEST) =< ?EPSILON),
?assert(abs(-100.0 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_NEGATIVE_INT_CONSTANT_TEST) =< ?EPSILON),
?assert(abs(9223372036854775807.0 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_LARGEST_INT_CONSTANT_TEST) =< ?EPSILON),
?assert(abs(-9223372036854775807.0 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_SMALLEST_INT_CONSTANT_TEST) =< ?EPSILON),
?assert(abs(3.14159265359 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_DOUBLE_WITH_MANY_DECIMALS_TEST) =< ?EPSILON),
?assert(abs(1000000.1 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_FRACTIONAL_DOUBLE_TEST) =< ?EPSILON),
?assert(abs(-1000000.1 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_NEGATIVE_FRACTIONAL_DOUBLE_TEST) =< ?EPSILON),
?assert(abs(1.7e+308 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_LARGE_DOUBLE_TEST) =< ?EPSILON),
?assert(abs(9223372036854775816.43 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_LARGE_FRACTIONAL_DOUBLE_TEST) =< ?EPSILON),
?assert(abs(-1.7e+308 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_SMALL_DOUBLE_TEST) =< ?EPSILON),
?assert(abs(-9223372036854775816.43 - ?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_NEGATIVE_BUT_LARGE_FRACTIONAL_DOUBLE_TEST) =< ?EPSILON),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_INT_CONSTANT_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_NEGATIVE_INT_CONSTANT_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_LARGEST_INT_CONSTANT_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_SMALLEST_INT_CONSTANT_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_DOUBLE_WITH_MANY_DECIMALS_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_FRACTIONAL_DOUBLE_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_NEGATIVE_FRACTIONAL_DOUBLE_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_LARGE_DOUBLE_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_LARGE_FRACTIONAL_DOUBLE_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_SMALL_DOUBLE_TEST)),
?assert(is_float(?DOUBLE_CONSTANTS_TEST_DOUBLE_ASSIGNED_TO_NEGATIVE_BUT_LARGE_FRACTIONAL_DOUBLE_TEST)).
rendered_double_list_test() ->
?assertEqual(12, length(?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)),
?assert(abs(1.0 - lists:nth(1, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(-100.0 - lists:nth(2, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(100.0 - lists:nth(3, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(9223372036854775807.0 - lists:nth(4, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(-9223372036854775807.0 - lists:nth(5, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(3.14159265359 - lists:nth(6, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(1000000.1 - lists:nth(7, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(-1000000.1 - lists:nth(8, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(1.7e+308 - lists:nth(9, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(-1.7e+308 - lists:nth(10, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(9223372036854775816.43 - lists:nth(11, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON),
?assert(abs(-9223372036854775816.43 - lists:nth(12, ?DOUBLE_CONSTANTS_TEST_DOUBLE_LIST_TEST)) =< ?EPSILON).
-endif. %% TEST

View File

@@ -31,14 +31,12 @@ install:
@echo '##############################################################'
check-local:
GOPATH=`pwd` $(GO) get golang.org/x/net/context
GOPATH=`pwd` $(GO) test -race ./thrift
clean-local:
$(RM) -rf pkg
all-local:
GOPATH=`pwd` $(GO) get golang.org/x/net/context
GOPATH=`pwd` $(GO) build ./thrift
EXTRA_DIST = \

View File

@@ -24,6 +24,8 @@ under the License.
Using Thrift with Go
====================
Thrift supports Go 1.7+
In following Go conventions, we recommend you use the 'go' tool to install
Thrift for go.

View File

@@ -17,10 +17,6 @@
# under the License.
#
if GOVERSION_LT_17
COMPILER_EXTRAFLAG=",legacy_context"
endif
THRIFTARGS = -out gopath/src/ --gen go:thrift_import=thrift$(COMPILER_EXTRAFLAG)
THRIFTTEST = $(top_srcdir)/test/ThriftTest.thrift
@@ -59,7 +55,6 @@ gopath: $(THRIFT) $(THRIFTTEST) \
$(THRIFT) $(THRIFTARGS) InitialismsTest.thrift
$(THRIFT) $(THRIFTARGS),read_write_private DontExportRWTest.thrift
$(THRIFT) $(THRIFTARGS),ignore_initialisms IgnoreInitialismsTest.thrift
GOPATH=`pwd`/gopath $(GO) get golang.org/x/net/context
GOPATH=`pwd`/gopath $(GO) get github.com/golang/mock/gomock || true
sed -i 's/\"context\"/\"golang.org\/x\/net\/context\"/g' gopath/src/github.com/golang/mock/gomock/controller.go || true
GOPATH=`pwd`/gopath $(GO) get github.com/golang/mock/gomock

View File

@@ -20,7 +20,6 @@
package dontexportrwtest
import (
"fmt"
"testing"
)
@@ -29,10 +28,10 @@ import (
func TestReadWriteMethodsArePrivate(t *testing.T) {
// This will only compile if read/write methods exist
s := NewTestStruct()
fmt.Sprintf("%v", s.read)
fmt.Sprintf("%v", s.write)
_ = s.read
_ = s.write
is := NewInnerStruct()
fmt.Sprintf("%v", is.read)
fmt.Sprintf("%v", is.write)
_ = is.read
_ = is.write
}

View File

@@ -20,6 +20,7 @@
package tests
import (
"context"
"errors"
"errortest"
"testing"
@@ -212,7 +213,7 @@ func prepareClientCallReply(protocol *MockTProtocol, failAt int, failWith error)
if failAt == 25 {
err = failWith
}
last = protocol.EXPECT().Flush().Return(err).After(last)
last = protocol.EXPECT().Flush(context.Background()).Return(err).After(last)
if failAt == 25 {
return true
}
@@ -414,6 +415,7 @@ func TestClientReportTTransportErrors(t *testing.T) {
client := errortest.NewErrorTestClient(thrift.NewTStandardClient(protocol, protocol))
_, retErr := client.TestStruct(defaultCtx, thing)
mockCtrl.Finish()
mockCtrl = gomock.NewController(t)
err2, ok := retErr.(thrift.TTransportException)
if !ok {
t.Fatal("Expected a TTrasportException")
@@ -446,6 +448,7 @@ func TestClientReportTTransportErrorsLegacy(t *testing.T) {
client := errortest.NewErrorTestClientProtocol(transport, protocol, protocol)
_, retErr := client.TestStruct(defaultCtx, thing)
mockCtrl.Finish()
mockCtrl = gomock.NewController(t)
err2, ok := retErr.(thrift.TTransportException)
if !ok {
t.Fatal("Expected a TTrasportException")
@@ -477,6 +480,7 @@ func TestClientReportTProtocolErrors(t *testing.T) {
client := errortest.NewErrorTestClient(thrift.NewTStandardClient(protocol, protocol))
_, retErr := client.TestStruct(defaultCtx, thing)
mockCtrl.Finish()
mockCtrl = gomock.NewController(t)
err2, ok := retErr.(thrift.TProtocolException)
if !ok {
t.Fatal("Expected a TProtocolException")
@@ -508,6 +512,7 @@ func TestClientReportTProtocolErrorsLegacy(t *testing.T) {
client := errortest.NewErrorTestClientProtocol(transport, protocol, protocol)
_, retErr := client.TestStruct(defaultCtx, thing)
mockCtrl.Finish()
mockCtrl = gomock.NewController(t)
err2, ok := retErr.(thrift.TProtocolException)
if !ok {
t.Fatal("Expected a TProtocolException")
@@ -532,7 +537,7 @@ func prepareClientCallException(protocol *MockTProtocol, failAt int, failWith er
last = protocol.EXPECT().WriteFieldStop().After(last)
last = protocol.EXPECT().WriteStructEnd().After(last)
last = protocol.EXPECT().WriteMessageEnd().After(last)
last = protocol.EXPECT().Flush().After(last)
last = protocol.EXPECT().Flush(context.Background()).After(last)
// Reading the exception, might fail.
if failAt == 0 {
@@ -628,6 +633,7 @@ func TestClientCallException(t *testing.T) {
client := errortest.NewErrorTestClient(thrift.NewTStandardClient(protocol, protocol))
_, retErr := client.TestString(defaultCtx, "test")
mockCtrl.Finish()
mockCtrl = gomock.NewController(t)
if !willComplete {
err2, ok := retErr.(thrift.TTransportException)
@@ -663,6 +669,7 @@ func TestClientCallExceptionLegacy(t *testing.T) {
client := errortest.NewErrorTestClientProtocol(transport, protocol, protocol)
_, retErr := client.TestString(defaultCtx, "test")
mockCtrl.Finish()
mockCtrl = gomock.NewController(t)
if !willComplete {
err2, ok := retErr.(thrift.TTransportException)
@@ -698,7 +705,7 @@ func TestClientSeqIdMismatch(t *testing.T) {
protocol.EXPECT().WriteFieldStop(),
protocol.EXPECT().WriteStructEnd(),
protocol.EXPECT().WriteMessageEnd(),
protocol.EXPECT().Flush(),
protocol.EXPECT().Flush(context.Background()),
protocol.EXPECT().ReadMessageBegin().Return("testString", thrift.REPLY, int32(2), nil),
)
@@ -729,7 +736,7 @@ func TestClientSeqIdMismatchLegeacy(t *testing.T) {
protocol.EXPECT().WriteFieldStop(),
protocol.EXPECT().WriteStructEnd(),
protocol.EXPECT().WriteMessageEnd(),
protocol.EXPECT().Flush(),
protocol.EXPECT().Flush(context.Background()),
protocol.EXPECT().ReadMessageBegin().Return("testString", thrift.REPLY, int32(2), nil),
)
@@ -758,7 +765,7 @@ func TestClientWrongMethodName(t *testing.T) {
protocol.EXPECT().WriteFieldStop(),
protocol.EXPECT().WriteStructEnd(),
protocol.EXPECT().WriteMessageEnd(),
protocol.EXPECT().Flush(),
protocol.EXPECT().Flush(context.Background()),
protocol.EXPECT().ReadMessageBegin().Return("unknown", thrift.REPLY, int32(1), nil),
)
@@ -789,7 +796,7 @@ func TestClientWrongMethodNameLegacy(t *testing.T) {
protocol.EXPECT().WriteFieldStop(),
protocol.EXPECT().WriteStructEnd(),
protocol.EXPECT().WriteMessageEnd(),
protocol.EXPECT().Flush(),
protocol.EXPECT().Flush(context.Background()),
protocol.EXPECT().ReadMessageBegin().Return("unknown", thrift.REPLY, int32(1), nil),
)
@@ -818,7 +825,7 @@ func TestClientWrongMessageType(t *testing.T) {
protocol.EXPECT().WriteFieldStop(),
protocol.EXPECT().WriteStructEnd(),
protocol.EXPECT().WriteMessageEnd(),
protocol.EXPECT().Flush(),
protocol.EXPECT().Flush(context.Background()),
protocol.EXPECT().ReadMessageBegin().Return("testString", thrift.INVALID_TMESSAGE_TYPE, int32(1), nil),
)
@@ -849,7 +856,7 @@ func TestClientWrongMessageTypeLegacy(t *testing.T) {
protocol.EXPECT().WriteFieldStop(),
protocol.EXPECT().WriteStructEnd(),
protocol.EXPECT().WriteMessageEnd(),
protocol.EXPECT().Flush(),
protocol.EXPECT().Flush(context.Background()),
protocol.EXPECT().ReadMessageBegin().Return("testString", thrift.INVALID_TMESSAGE_TYPE, int32(1), nil),
)

View File

@@ -1,5 +1,3 @@
// +build !go1.7
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -19,8 +17,10 @@
* under the License.
*/
package thrift
package tests
import "golang.org/x/net/context"
import (
"context"
)
var defaultCtx = context.Background()

View File

@@ -20,6 +20,7 @@
package tests
import (
"context"
"multiplexedprotocoltest"
"net"
"testing"
@@ -36,6 +37,18 @@ func FindAvailableTCPServerPort() net.Addr {
}
}
type FirstImpl struct{}
func (f *FirstImpl) ReturnOne(ctx context.Context) (r int64, err error) {
return 1, nil
}
type SecondImpl struct{}
func (s *SecondImpl) ReturnTwo(ctx context.Context) (r int64, err error) {
return 2, nil
}
func createTransport(addr net.Addr) (thrift.TTransport, error) {
socket := thrift.NewTSocketFromAddrTimeout(addr, TIMEOUT)
transport := thrift.NewTFramedTransport(socket)

View File

@@ -20,6 +20,8 @@
package tests
import (
"context"
"fmt"
"net"
"onewaytest"
"testing"
@@ -36,6 +38,12 @@ func findPort() net.Addr {
}
}
type impl struct{}
func (i *impl) Hi(ctx context.Context, in int64, s string) (err error) { fmt.Println("Hi!"); return }
func (i *impl) Emptyfunc(ctx context.Context) (err error) { return }
func (i *impl) EchoInt(ctx context.Context, param int64) (r int64, err error) { return param, nil }
const TIMEOUT = time.Second
var addr net.Addr

View File

@@ -23,6 +23,7 @@
package tests
import (
"context"
thrift "thrift"
gomock "github.com/golang/mock/gomock"
@@ -49,13 +50,13 @@ func (_m *MockTProtocol) EXPECT() *_MockTProtocolRecorder {
return _m.recorder
}
func (_m *MockTProtocol) Flush() error {
func (_m *MockTProtocol) Flush(ctx context.Context) error {
ret := _m.ctrl.Call(_m, "Flush")
ret0, _ := ret[0].(error)
return ret0
}
func (_mr *_MockTProtocolRecorder) Flush() *gomock.Call {
func (_mr *_MockTProtocolRecorder) Flush(ctx context.Context) *gomock.Call {
return _mr.mock.ctrl.RecordCall(_mr.mock, "Flush")
}

View File

@@ -40,6 +40,7 @@ func TestStructReadRequiredFields(t *testing.T) {
err := testStruct.Read(protocol)
mockCtrl.Finish()
mockCtrl = gomock.NewController(t)
if err == nil {
t.Fatal("Expected read to fail")
}
@@ -63,6 +64,7 @@ func TestStructReadRequiredFields(t *testing.T) {
err = testStruct.Read(protocol)
mockCtrl.Finish()
mockCtrl = gomock.NewController(t)
if err == nil {
t.Fatal("Expected read to fail")
}

View File

@@ -1,5 +1,3 @@
// +build !go1.7
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -22,12 +20,11 @@
package tests
import (
"context"
"errors"
"thrift"
"thrifttest"
"time"
"golang.org/x/net/context"
)
type SecondServiceHandler struct {

View File

@@ -1,212 +0,0 @@
// +build go1.7
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* 'License'); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* 'AS IS' BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
* KIND, either express or implied. See the License for the
* specific language governing permissions and limitations
* under the License.
*/
package tests
import (
"context"
"errors"
"thrift"
"thrifttest"
"time"
)
type SecondServiceHandler struct {
}
func NewSecondServiceHandler() *SecondServiceHandler {
return &SecondServiceHandler{}
}
func (p *SecondServiceHandler) BlahBlah(ctx context.Context) (err error) {
return nil
}
func (p *SecondServiceHandler) SecondtestString(ctx context.Context, thing string) (r string, err error) {
return thing, nil
}
type ThriftTestHandler struct {
}
func NewThriftTestHandler() *ThriftTestHandler {
return &ThriftTestHandler{}
}
func (p *ThriftTestHandler) TestVoid(ctx context.Context) (err error) {
return nil
}
func (p *ThriftTestHandler) TestString(ctx context.Context, thing string) (r string, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestBool(ctx context.Context, thing bool) (r bool, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestByte(ctx context.Context, thing int8) (r int8, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestI32(ctx context.Context, thing int32) (r int32, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestI64(ctx context.Context, thing int64) (r int64, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestDouble(ctx context.Context, thing float64) (r float64, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestBinary(ctx context.Context, thing []byte) (r []byte, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestStruct(ctx context.Context, thing *thrifttest.Xtruct) (r *thrifttest.Xtruct, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestNest(ctx context.Context, thing *thrifttest.Xtruct2) (r *thrifttest.Xtruct2, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestMap(ctx context.Context, thing map[int32]int32) (r map[int32]int32, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestStringMap(ctx context.Context, thing map[string]string) (r map[string]string, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestSet(ctx context.Context, thing []int32) (r []int32, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestList(ctx context.Context, thing []int32) (r []int32, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestEnum(ctx context.Context, thing thrifttest.Numberz) (r thrifttest.Numberz, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestTypedef(ctx context.Context, thing thrifttest.UserId) (r thrifttest.UserId, err error) {
return thing, nil
}
func (p *ThriftTestHandler) TestMapMap(ctx context.Context, hello int32) (r map[int32]map[int32]int32, err error) {
r = make(map[int32]map[int32]int32)
pos := make(map[int32]int32)
neg := make(map[int32]int32)
for i := int32(1); i < 5; i++ {
pos[i] = i
neg[-i] = -i
}
r[4] = pos
r[-4] = neg
return r, nil
}
func (p *ThriftTestHandler) TestInsanity(ctx context.Context, argument *thrifttest.Insanity) (r map[thrifttest.UserId]map[thrifttest.Numberz]*thrifttest.Insanity, err error) {
hello := thrifttest.NewXtruct()
hello.StringThing = "Hello2"
hello.ByteThing = 2
hello.I32Thing = 2
hello.I64Thing = 2
goodbye := thrifttest.NewXtruct()
goodbye.StringThing = "Goodbye4"
goodbye.ByteThing = 4
goodbye.I32Thing = 4
goodbye.I64Thing = 4
crazy := thrifttest.NewInsanity()
crazy.UserMap = make(map[thrifttest.Numberz]thrifttest.UserId)
crazy.UserMap[thrifttest.Numberz_EIGHT] = 8
crazy.UserMap[thrifttest.Numberz_FIVE] = 5
crazy.Xtructs = []*thrifttest.Xtruct{goodbye, hello}
first_map := make(map[thrifttest.Numberz]*thrifttest.Insanity)
second_map := make(map[thrifttest.Numberz]*thrifttest.Insanity)
first_map[thrifttest.Numberz_TWO] = crazy
first_map[thrifttest.Numberz_THREE] = crazy
looney := thrifttest.NewInsanity()
second_map[thrifttest.Numberz_SIX] = looney
var insane = make(map[thrifttest.UserId]map[thrifttest.Numberz]*thrifttest.Insanity)
insane[1] = first_map
insane[2] = second_map
return insane, nil
}
func (p *ThriftTestHandler) TestMulti(ctx context.Context, arg0 int8, arg1 int32, arg2 int64, arg3 map[int16]string, arg4 thrifttest.Numberz, arg5 thrifttest.UserId) (r *thrifttest.Xtruct, err error) {
r = thrifttest.NewXtruct()
r.StringThing = "Hello2"
r.ByteThing = arg0
r.I32Thing = arg1
r.I64Thing = arg2
return r, nil
}
func (p *ThriftTestHandler) TestException(ctx context.Context, arg string) (err error) {
if arg == "Xception" {
x := thrifttest.NewXception()
x.ErrorCode = 1001
x.Message = arg
return x
} else if arg == "TException" {
return thrift.TException(errors.New(arg))
} else {
return nil
}
}
func (p *ThriftTestHandler) TestMultiException(ctx context.Context, arg0 string, arg1 string) (r *thrifttest.Xtruct, err error) {
if arg0 == "Xception" {
x := thrifttest.NewXception()
x.ErrorCode = 1001
x.Message = "This is an Xception"
return nil, x
} else if arg0 == "Xception2" {
x2 := thrifttest.NewXception2()
x2.ErrorCode = 2002
x2.StructThing = thrifttest.NewXtruct()
x2.StructThing.StringThing = "This is an Xception2"
return nil, x2
}
res := thrifttest.NewXtruct()
res.StringThing = arg1
return res, nil
}
func (p *ThriftTestHandler) TestOneway(ctx context.Context, secondsToSleep int32) (err error) {
time.Sleep(time.Second * time.Duration(secondsToSleep))
return nil
}

View File

@@ -29,13 +29,13 @@ func TestTApplicationException(t *testing.T) {
t.Fatalf("Expected empty string for exception but found '%s'", exc.Error())
}
if exc.TypeId() != UNKNOWN_APPLICATION_EXCEPTION {
t.Fatalf("Expected type UNKNOWN for exception but found '%s'", exc.TypeId())
t.Fatalf("Expected type UNKNOWN for exception but found '%v'", exc.TypeId())
}
exc = NewTApplicationException(WRONG_METHOD_NAME, "junk_method")
if exc.Error() != "junk_method" {
t.Fatalf("Expected 'junk_method' for exception but found '%s'", exc.Error())
}
if exc.TypeId() != WRONG_METHOD_NAME {
t.Fatalf("Expected type WRONG_METHOD_NAME for exception but found '%s'", exc.TypeId())
t.Fatalf("Expected type WRONG_METHOD_NAME for exception but found '%v'", exc.TypeId())
}
}

View File

@@ -21,6 +21,7 @@ package thrift
import (
"bytes"
"context"
"encoding/binary"
"errors"
"fmt"
@@ -457,8 +458,8 @@ func (p *TBinaryProtocol) ReadBinary() ([]byte, error) {
return buf, NewTProtocolException(err)
}
func (p *TBinaryProtocol) Flush() (err error) {
return NewTProtocolException(p.trans.Flush())
func (p *TBinaryProtocol) Flush(ctx context.Context) (err error) {
return NewTProtocolException(p.trans.Flush(ctx))
}
func (p *TBinaryProtocol) Skip(fieldType TType) (err error) {

View File

@@ -21,6 +21,7 @@ package thrift
import (
"bufio"
"context"
)
type TBufferedTransportFactory struct {
@@ -78,12 +79,12 @@ func (p *TBufferedTransport) Write(b []byte) (int, error) {
return n, err
}
func (p *TBufferedTransport) Flush() error {
func (p *TBufferedTransport) Flush(ctx context.Context) error {
if err := p.ReadWriter.Flush(); err != nil {
p.ReadWriter.Writer.Reset(p.tp)
return err
}
return p.tp.Flush()
return p.tp.Flush(ctx)
}
func (p *TBufferedTransport) RemainingBytes() (num_bytes uint64) {

View File

@@ -1,6 +1,13 @@
package thrift
import "fmt"
import (
"context"
"fmt"
)
type TClient interface {
Call(ctx context.Context, method string, args, result TStruct) error
}
type TStandardClient struct {
seqId int32
@@ -16,7 +23,7 @@ func NewTStandardClient(inputProtocol, outputProtocol TProtocol) *TStandardClien
}
}
func (p *TStandardClient) Send(oprot TProtocol, seqId int32, method string, args TStruct) error {
func (p *TStandardClient) Send(ctx context.Context, oprot TProtocol, seqId int32, method string, args TStruct) error {
if err := oprot.WriteMessageBegin(method, CALL, seqId); err != nil {
return err
}
@@ -26,7 +33,7 @@ func (p *TStandardClient) Send(oprot TProtocol, seqId int32, method string, args
if err := oprot.WriteMessageEnd(); err != nil {
return err
}
return oprot.Flush()
return oprot.Flush(ctx)
}
func (p *TStandardClient) Recv(iprot TProtocol, seqId int32, method string, result TStruct) error {
@@ -61,11 +68,11 @@ func (p *TStandardClient) Recv(iprot TProtocol, seqId int32, method string, resu
return iprot.ReadMessageEnd()
}
func (p *TStandardClient) call(method string, args, result TStruct) error {
func (p *TStandardClient) Call(ctx context.Context, method string, args, result TStruct) error {
p.seqId++
seqId := p.seqId
if err := p.Send(p.oprot, seqId, method, args); err != nil {
if err := p.Send(ctx, p.oprot, seqId, method, args); err != nil {
return err
}

View File

@@ -1,13 +0,0 @@
// +build go1.7
package thrift
import "context"
type TClient interface {
Call(ctx context.Context, method string, args, result TStruct) error
}
func (p *TStandardClient) Call(ctx context.Context, method string, args, result TStruct) error {
return p.call(method, args, result)
}

View File

@@ -1,13 +0,0 @@
// +build !go1.7
package thrift
import "golang.org/x/net/context"
type TClient interface {
Call(ctx context.Context, method string, args, result TStruct) error
}
func (p *TStandardClient) Call(ctx context.Context, method string, args, result TStruct) error {
return p.call(method, args, result)
}

View File

@@ -1,5 +1,3 @@
// +build go1.7
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file

View File

@@ -20,6 +20,7 @@
package thrift
import (
"context"
"encoding/binary"
"fmt"
"io"
@@ -599,8 +600,8 @@ func (p *TCompactProtocol) ReadBinary() (value []byte, err error) {
return buf, NewTProtocolException(e)
}
func (p *TCompactProtocol) Flush() (err error) {
return NewTProtocolException(p.trans.Flush())
func (p *TCompactProtocol) Flush(ctx context.Context) (err error) {
return NewTProtocolException(p.trans.Flush(ctx))
}
func (p *TCompactProtocol) Skip(fieldType TType) (err error) {
@@ -806,7 +807,7 @@ func (p *TCompactProtocol) getTType(t tCompactType) (TType, error) {
case COMPACT_STRUCT:
return STRUCT, nil
}
return STOP, TException(fmt.Errorf("don't know what type: %s", t&0x0f))
return STOP, TException(fmt.Errorf("don't know what type: %v", t&0x0f))
}
// Given a TType value, find the appropriate TCompactProtocol.Types constant.

View File

@@ -1,5 +1,3 @@
// +build go1.7
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file

View File

@@ -20,6 +20,7 @@
package thrift
import (
"context"
"log"
)
@@ -258,8 +259,8 @@ func (tdp *TDebugProtocol) Skip(fieldType TType) (err error) {
log.Printf("%sSkip(fieldType=%#v) (err=%#v)", tdp.LogPrefix, fieldType, err)
return
}
func (tdp *TDebugProtocol) Flush() (err error) {
err = tdp.Delegate.Flush()
func (tdp *TDebugProtocol) Flush(ctx context.Context) (err error) {
err = tdp.Delegate.Flush(ctx)
log.Printf("%sFlush() (err=%#v)", tdp.LogPrefix, err)
return
}

View File

@@ -22,6 +22,7 @@ package thrift
import (
"bufio"
"bytes"
"context"
"encoding/binary"
"fmt"
"io"
@@ -135,7 +136,7 @@ func (p *TFramedTransport) WriteString(s string) (n int, err error) {
return p.buf.WriteString(s)
}
func (p *TFramedTransport) Flush() error {
func (p *TFramedTransport) Flush(ctx context.Context) error {
size := p.buf.Len()
buf := p.buffer[:4]
binary.BigEndian.PutUint32(buf, uint32(size))
@@ -151,7 +152,7 @@ func (p *TFramedTransport) Flush() error {
return NewTTransportExceptionFromError(err)
}
}
err = p.transport.Flush()
err = p.transport.Flush(ctx)
return NewTTransportExceptionFromError(err)
}

View File

@@ -21,6 +21,7 @@ package thrift
import (
"bytes"
"context"
"io"
"io/ioutil"
"net/http"
@@ -181,7 +182,7 @@ func (p *THttpClient) WriteString(s string) (n int, err error) {
return p.requestBuffer.WriteString(s)
}
func (p *THttpClient) Flush() error {
func (p *THttpClient) Flush(ctx context.Context) error {
// Close any previous response body to avoid leaking connections.
p.closeResponse()
@@ -190,6 +191,9 @@ func (p *THttpClient) Flush() error {
return NewTTransportExceptionFromError(err)
}
req.Header = p.header
if ctx != nil {
req = req.WithContext(ctx)
}
response, err := p.client.Do(req)
if err != nil {
return NewTTransportExceptionFromError(err)

View File

@@ -26,6 +26,18 @@ import (
"strings"
)
// NewThriftHandlerFunc is a function that create a ready to use Apache Thrift Handler function
func NewThriftHandlerFunc(processor TProcessor,
inPfactory, outPfactory TProtocolFactory) func(w http.ResponseWriter, r *http.Request) {
return gz(func(w http.ResponseWriter, r *http.Request) {
w.Header().Add("Content-Type", "application/x-thrift")
transport := NewStreamTransport(r.Body, w)
processor.Process(r.Context(), inPfactory.GetProtocol(transport), outPfactory.GetProtocol(transport))
})
}
// gz transparently compresses the HTTP response if the client supports it.
func gz(handler http.HandlerFunc) http.HandlerFunc {
return func(w http.ResponseWriter, r *http.Request) {

Some files were not shown because too many files have changed in this diff Show More