Skip to content
Projects
Groups
Snippets
Help
Loading...
Sign in / Register
Toggle navigation
C
capnproto
Project
Project
Details
Activity
Cycle Analytics
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Charts
Issues
0
Issues
0
List
Board
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Charts
Packages
Packages
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Charts
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
submodule
capnproto
Commits
7260fe75
Unverified
Commit
7260fe75
authored
Oct 08, 2018
by
Kenton Varda
Committed by
GitHub
Oct 08, 2018
Browse files
Options
Browse Files
Download
Plain Diff
Merge pull request #748 from capnproto/json-rpc
Implement JSON-RPC adapter to Cap'n Proto interfaces.
parents
9f31cd49
5f422f65
Hide whitespace changes
Inline
Side-by-side
Showing
11 changed files
with
886 additions
and
26 deletions
+886
-26
json-rpc-test.c++
c++/src/capnp/compat/json-rpc-test.c++
+100
-0
json-rpc.c++
c++/src/capnp/compat/json-rpc.c++
+341
-0
json-rpc.capnp
c++/src/capnp/compat/json-rpc.capnp
+43
-0
json-rpc.h
c++/src/capnp/compat/json-rpc.h
+112
-0
json.capnp
c++/src/capnp/compat/json.capnp
+8
-5
json.h
c++/src/capnp/compat/json.h
+6
-0
dynamic.h
c++/src/capnp/dynamic.h
+3
-0
serialize-async.c++
c++/src/capnp/serialize-async.c++
+5
-4
http-test.c++
c++/src/kj/compat/http-test.c++
+135
-0
http.c++
c++/src/kj/compat/http.c++
+68
-17
http.h
c++/src/kj/compat/http.h
+65
-0
No files found.
c++/src/capnp/compat/json-rpc-test.c++
0 → 100644
View file @
7260fe75
// Copyright (c) 2018 Kenton Varda and contributors
// Licensed under the MIT License:
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
#include "json-rpc.h"
#include <kj/test.h>
#include <capnp/test-util.h>
namespace
capnp
{
namespace
_
{
// private
namespace
{
KJ_TEST
(
"json-rpc basics"
)
{
auto
io
=
kj
::
setupAsyncIo
();
auto
pipe
=
kj
::
newTwoWayPipe
();
JsonRpc
::
ContentLengthTransport
clientTransport
(
*
pipe
.
ends
[
0
]);
JsonRpc
::
ContentLengthTransport
serverTransport
(
*
pipe
.
ends
[
1
]);
int
callCount
=
0
;
JsonRpc
client
(
clientTransport
);
JsonRpc
server
(
serverTransport
,
toDynamic
(
kj
::
heap
<
TestInterfaceImpl
>
(
callCount
)));
auto
cap
=
client
.
getPeer
<
test
::
TestInterface
>
();
auto
req
=
cap
.
fooRequest
();
req
.
setI
(
123
);
req
.
setJ
(
true
);
auto
resp
=
req
.
send
().
wait
(
io
.
waitScope
);
KJ_EXPECT
(
resp
.
getX
()
==
"foo"
);
KJ_EXPECT
(
callCount
==
1
);
}
KJ_TEST
(
"json-rpc error"
)
{
auto
io
=
kj
::
setupAsyncIo
();
auto
pipe
=
kj
::
newTwoWayPipe
();
JsonRpc
::
ContentLengthTransport
clientTransport
(
*
pipe
.
ends
[
0
]);
JsonRpc
::
ContentLengthTransport
serverTransport
(
*
pipe
.
ends
[
1
]);
int
callCount
=
0
;
JsonRpc
client
(
clientTransport
);
JsonRpc
server
(
serverTransport
,
toDynamic
(
kj
::
heap
<
TestInterfaceImpl
>
(
callCount
)));
auto
cap
=
client
.
getPeer
<
test
::
TestInterface
>
();
KJ_EXPECT_THROW_MESSAGE
(
"Method not implemented"
,
cap
.
barRequest
().
send
().
wait
(
io
.
waitScope
));
}
KJ_TEST
(
"json-rpc multiple calls"
)
{
auto
io
=
kj
::
setupAsyncIo
();
auto
pipe
=
kj
::
newTwoWayPipe
();
JsonRpc
::
ContentLengthTransport
clientTransport
(
*
pipe
.
ends
[
0
]);
JsonRpc
::
ContentLengthTransport
serverTransport
(
*
pipe
.
ends
[
1
]);
int
callCount
=
0
;
JsonRpc
client
(
clientTransport
);
JsonRpc
server
(
serverTransport
,
toDynamic
(
kj
::
heap
<
TestInterfaceImpl
>
(
callCount
)));
auto
cap
=
client
.
getPeer
<
test
::
TestInterface
>
();
auto
req1
=
cap
.
fooRequest
();
req1
.
setI
(
123
);
req1
.
setJ
(
true
);
auto
promise1
=
req1
.
send
();
auto
req2
=
cap
.
bazRequest
();
initTestMessage
(
req2
.
initS
());
auto
promise2
=
req2
.
send
();
auto
resp1
=
promise1
.
wait
(
io
.
waitScope
);
KJ_EXPECT
(
resp1
.
getX
()
==
"foo"
);
auto
resp2
=
promise2
.
wait
(
io
.
waitScope
);
KJ_EXPECT
(
callCount
==
2
);
}
}
// namespace
}
// namespace _ (private)
}
// namespace capnp
c++/src/capnp/compat/json-rpc.c++
0 → 100644
View file @
7260fe75
// Copyright (c) 2018 Kenton Varda and contributors
// Licensed under the MIT License:
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
#include "json-rpc.h"
#include <kj/compat/http.h>
#include <capnp/compat/json-rpc.capnp.h>
namespace
capnp
{
static
constexpr
uint64_t
JSON_NAME_ANNOTATION_ID
=
0xfa5b1fd61c2e7c3dull
;
static
constexpr
uint64_t
JSON_NOTIFICATION_ANNOTATION_ID
=
0xa0a054dea32fd98cull
;
class
JsonRpc
::
CapabilityImpl
final
:
public
DynamicCapability
::
Server
{
public
:
CapabilityImpl
(
JsonRpc
&
parent
,
InterfaceSchema
schema
)
:
DynamicCapability
::
Server
(
schema
),
parent
(
parent
)
{}
kj
::
Promise
<
void
>
call
(
InterfaceSchema
::
Method
method
,
CallContext
<
DynamicStruct
,
DynamicStruct
>
context
)
override
{
auto
proto
=
method
.
getProto
();
bool
isNotification
=
false
;
kj
::
StringPtr
name
=
proto
.
getName
();
for
(
auto
annotation
:
proto
.
getAnnotations
())
{
switch
(
annotation
.
getId
())
{
case
JSON_NAME_ANNOTATION_ID
:
name
=
annotation
.
getValue
().
getText
();
break
;
case
JSON_NOTIFICATION_ANNOTATION_ID
:
isNotification
=
true
;
break
;
}
}
capnp
::
MallocMessageBuilder
message
;
auto
value
=
message
.
getRoot
<
json
::
Value
>
();
auto
list
=
value
.
initObject
(
3
+
!
isNotification
);
uint
index
=
0
;
auto
jsonrpc
=
list
[
index
++
];
jsonrpc
.
setName
(
"jsonrpc"
);
jsonrpc
.
initValue
().
setString
(
"2.0"
);
uint
callId
=
parent
.
callCount
++
;
if
(
!
isNotification
)
{
auto
id
=
list
[
index
++
];
id
.
setName
(
"id"
);
id
.
initValue
().
setNumber
(
callId
);
}
auto
methodName
=
list
[
index
++
];
methodName
.
setName
(
"method"
);
methodName
.
initValue
().
setString
(
name
);
auto
params
=
list
[
index
++
];
params
.
setName
(
"params"
);
parent
.
codec
.
encode
(
context
.
getParams
(),
params
.
initValue
());
auto
writePromise
=
parent
.
queueWrite
(
parent
.
codec
.
encode
(
value
));
if
(
isNotification
)
{
auto
sproto
=
context
.
getResultsType
().
getProto
().
getStruct
();
MessageSize
size
{
sproto
.
getDataWordCount
(),
sproto
.
getPointerCount
()
};
context
.
initResults
(
size
);
return
kj
::
mv
(
writePromise
);
}
else
{
auto
paf
=
kj
::
newPromiseAndFulfiller
<
void
>
();
parent
.
awaitedResponses
.
insert
(
callId
,
AwaitedResponse
{
context
,
kj
::
mv
(
paf
.
fulfiller
)
});
auto
promise
=
writePromise
.
then
([
p
=
kj
::
mv
(
paf
.
promise
)]()
mutable
{
return
kj
::
mv
(
p
);
});
auto
&
parentRef
=
parent
;
return
promise
.
attach
(
kj
::
defer
([
&
parentRef
,
callId
]()
{
parentRef
.
awaitedResponses
.
erase
(
callId
);
}));
}
}
private
:
JsonRpc
&
parent
;
};
JsonRpc
::
JsonRpc
(
Transport
&
transport
,
DynamicCapability
::
Client
interface
)
:
JsonRpc
(
transport
,
kj
::
mv
(
interface
),
kj
::
newPromiseAndFulfiller
<
void
>
())
{}
JsonRpc
::
JsonRpc
(
Transport
&
transport
,
DynamicCapability
::
Client
interfaceParam
,
kj
::
PromiseFulfillerPair
<
void
>
paf
)
:
transport
(
transport
),
interface
(
kj
::
mv
(
interfaceParam
)),
errorPromise
(
paf
.
promise
.
fork
()),
errorFulfiller
(
kj
::
mv
(
paf
.
fulfiller
)),
readTask
(
readLoop
().
eagerlyEvaluate
([
this
](
kj
::
Exception
&&
e
)
{
errorFulfiller
->
reject
(
kj
::
mv
(
e
));
})),
tasks
(
*
this
)
{
codec
.
handleByAnnotation
(
interface
.
getSchema
());
codec
.
handleByAnnotation
<
json
::
RpcMessage
>
();
for
(
auto
method
:
interface
.
getSchema
().
getMethods
())
{
auto
proto
=
method
.
getProto
();
kj
::
StringPtr
name
=
proto
.
getName
();
for
(
auto
annotation
:
proto
.
getAnnotations
())
{
switch
(
annotation
.
getId
())
{
case
JSON_NAME_ANNOTATION_ID
:
name
=
annotation
.
getValue
().
getText
();
break
;
}
}
methodMap
.
insert
(
name
,
method
);
}
}
DynamicCapability
::
Client
JsonRpc
::
getPeer
(
InterfaceSchema
schema
)
{
codec
.
handleByAnnotation
(
interface
.
getSchema
());
return
kj
::
heap
<
CapabilityImpl
>
(
*
this
,
schema
);
}
static
kj
::
HttpHeaderTable
&
staticHeaderTable
()
{
static
kj
::
HttpHeaderTable
HEADER_TABLE
;
return
HEADER_TABLE
;
}
kj
::
Promise
<
void
>
JsonRpc
::
queueWrite
(
kj
::
String
text
)
{
auto
fork
=
writeQueue
.
then
([
this
,
text
=
kj
::
mv
(
text
)]()
mutable
{
auto
promise
=
transport
.
send
(
text
);
return
promise
.
attach
(
kj
::
mv
(
text
));
}).
eagerlyEvaluate
([
this
](
kj
::
Exception
&&
e
)
{
errorFulfiller
->
reject
(
kj
::
mv
(
e
));
}).
fork
();
writeQueue
=
fork
.
addBranch
();
return
fork
.
addBranch
();
}
void
JsonRpc
::
queueError
(
kj
::
Maybe
<
json
::
Value
::
Reader
>
id
,
int
code
,
kj
::
StringPtr
message
)
{
MallocMessageBuilder
capnpMessage
;
auto
jsonResponse
=
capnpMessage
.
getRoot
<
json
::
RpcMessage
>
();
jsonResponse
.
setJsonrpc
(
"2.0"
);
KJ_IF_MAYBE
(
i
,
id
)
{
jsonResponse
.
setId
(
*
i
);
}
else
{
jsonResponse
.
initId
().
setNull
();
}
auto
error
=
jsonResponse
.
initError
();
error
.
setCode
(
code
);
error
.
setMessage
(
message
);
// OK to discard result of queueWrite() since it's just one branch of a fork.
queueWrite
(
codec
.
encode
(
jsonResponse
));
}
kj
::
Promise
<
void
>
JsonRpc
::
readLoop
()
{
return
transport
.
receive
().
then
([
this
](
kj
::
String
message
)
->
kj
::
Promise
<
void
>
{
MallocMessageBuilder
capnpMessage
;
auto
rpcMessageBuilder
=
capnpMessage
.
getRoot
<
json
::
RpcMessage
>
();
KJ_IF_MAYBE
(
exception
,
kj
::
runCatchingExceptions
([
&
]()
{
codec
.
decode
(
message
,
rpcMessageBuilder
);
}))
{
queueError
(
nullptr
,
-
32700
,
kj
::
str
(
"Parse error: "
,
exception
->
getDescription
()));
return
readLoop
();
}
KJ_CONTEXT
(
"decoding JSON-RPC message"
,
message
);
auto
rpcMessage
=
rpcMessageBuilder
.
asReader
();
if
(
!
rpcMessage
.
hasJsonrpc
())
{
queueError
(
nullptr
,
-
32700
,
kj
::
str
(
"Missing 'jsonrpc' field."
));
return
readLoop
();
}
else
if
(
rpcMessage
.
getJsonrpc
()
!=
"2.0"
)
{
queueError
(
nullptr
,
-
32700
,
kj
::
str
(
"Unknown JSON-RPC version. This peer implements version '2.0'."
));
return
readLoop
();
}
switch
(
rpcMessage
.
which
())
{
case
json
:
:
RpcMessage
::
NONE
:
queueError
(
nullptr
,
-
32700
,
kj
::
str
(
"message has none of params, result, or error"
));
break
;
case
json
:
:
RpcMessage
::
PARAMS
:
{
// a call
auto
schema
=
interface
.
getSchema
();
KJ_IF_MAYBE
(
method
,
schema
.
findMethodByName
(
rpcMessage
.
getMethod
()))
{
auto
req
=
interface
.
newRequest
(
*
method
);
KJ_IF_MAYBE
(
exception
,
kj
::
runCatchingExceptions
([
&
]()
{
codec
.
decode
(
rpcMessage
.
getParams
(),
req
);
}))
{
kj
::
Maybe
<
JsonValue
::
Reader
>
id
;
if
(
rpcMessage
.
hasId
())
id
=
rpcMessage
.
getId
();
queueError
(
id
,
-
32602
,
kj
::
str
(
"Type error in method params: "
,
exception
->
getDescription
()));
break
;
}
if
(
rpcMessage
.
hasId
())
{
auto
id
=
rpcMessage
.
getId
();
auto
idCopy
=
kj
::
heapArray
<
word
>
(
id
.
totalSize
().
wordCount
+
1
);
memset
(
idCopy
.
begin
(),
0
,
idCopy
.
asBytes
().
size
());
copyToUnchecked
(
id
,
idCopy
);
auto
idPtr
=
readMessageUnchecked
<
json
::
Value
>
(
idCopy
.
begin
());
auto
promise
=
req
.
send
()
.
then
([
this
,
idPtr
](
Response
<
DynamicStruct
>
response
)
mutable
{
MallocMessageBuilder
capnpMessage
;
auto
jsonResponse
=
capnpMessage
.
getRoot
<
json
::
RpcMessage
>
();
jsonResponse
.
setJsonrpc
(
"2.0"
);
jsonResponse
.
setId
(
idPtr
);
codec
.
encode
(
DynamicStruct
::
Reader
(
response
),
jsonResponse
.
initResult
());
return
queueWrite
(
codec
.
encode
(
jsonResponse
));
},
[
this
,
idPtr
](
kj
::
Exception
&&
e
)
{
MallocMessageBuilder
capnpMessage
;
auto
jsonResponse
=
capnpMessage
.
getRoot
<
json
::
RpcMessage
>
();
jsonResponse
.
setJsonrpc
(
"2.0"
);
jsonResponse
.
setId
(
idPtr
);
auto
error
=
jsonResponse
.
initError
();
switch
(
e
.
getType
())
{
case
kj
:
:
Exception
::
Type
::
FAILED
:
error
.
setCode
(
-
32000
);
break
;
case
kj
:
:
Exception
::
Type
::
DISCONNECTED
:
error
.
setCode
(
-
32001
);
break
;
case
kj
:
:
Exception
::
Type
::
OVERLOADED
:
error
.
setCode
(
-
32002
);
break
;
case
kj
:
:
Exception
::
Type
::
UNIMPLEMENTED
:
error
.
setCode
(
-
32601
);
// method not found
break
;
}
error
.
setMessage
(
e
.
getDescription
());
return
queueWrite
(
codec
.
encode
(
jsonResponse
));
});
tasks
.
add
(
promise
.
attach
(
kj
::
mv
(
idCopy
)));
}
else
{
// No 'id', so this is a notification.
tasks
.
add
(
req
.
send
().
ignoreResult
().
catch_
([](
kj
::
Exception
&&
exception
)
{
if
(
exception
.
getType
()
!=
kj
::
Exception
::
Type
::
UNIMPLEMENTED
)
{
KJ_LOG
(
ERROR
,
"JSON-RPC notification threw exception into the abyss"
,
exception
);
}
}));
}
}
else
{
if
(
rpcMessage
.
hasId
())
{
queueError
(
rpcMessage
.
getId
(),
-
32601
,
"Method not found"
);
}
else
{
// Ignore notification for unknown method.
}
}
break
;
}
case
json
:
:
RpcMessage
::
RESULT
:
{
auto
id
=
rpcMessage
.
getId
();
if
(
!
id
.
isNumber
())
{
// JSON-RPC doesn't define what to do if receiving a response with an invalid id.
KJ_LOG
(
ERROR
,
"JSON-RPC response has invalid ID"
);
}
else
KJ_IF_MAYBE
(
awaited
,
awaitedResponses
.
find
((
uint
)
id
.
getNumber
()))
{
KJ_IF_MAYBE
(
exception
,
kj
::
runCatchingExceptions
([
&
]()
{
codec
.
decode
(
rpcMessage
.
getResult
(),
awaited
->
context
.
getResults
());
awaited
->
fulfiller
->
fulfill
();
}))
{
// Errors always propagate from callee to caller, so we don't want to throw this error
// back to the server.
awaited
->
fulfiller
->
reject
(
kj
::
mv
(
*
exception
));
}
}
else
{
// Probably, this is the response to a call that was canceled.
}
break
;
}
case
json
:
:
RpcMessage
::
ERROR
:
{
auto
id
=
rpcMessage
.
getId
();
if
(
id
.
isNull
())
{
// Error message will be logged by KJ_CONTEXT, above.
KJ_LOG
(
ERROR
,
"peer reports JSON-RPC protocol error"
);
}
else
if
(
!
id
.
isNumber
())
{
// JSON-RPC doesn't define what to do if receiving a response with an invalid id.
KJ_LOG
(
ERROR
,
"JSON-RPC response has invalid ID"
);
}
else
KJ_IF_MAYBE
(
awaited
,
awaitedResponses
.
find
((
uint
)
id
.
getNumber
()))
{
auto
error
=
rpcMessage
.
getError
();
auto
code
=
error
.
getCode
();
kj
::
Exception
::
Type
type
=
code
==
-
32601
?
kj
::
Exception
::
Type
::
UNIMPLEMENTED
:
kj
::
Exception
::
Type
::
FAILED
;
awaited
->
fulfiller
->
reject
(
kj
::
Exception
(
type
,
__FILE__
,
__LINE__
,
kj
::
str
(
error
.
getMessage
())));
}
else
{
// Probably, this is the response to a call that was canceled.
}
break
;
}
}
return
readLoop
();
});
}
void
JsonRpc
::
taskFailed
(
kj
::
Exception
&&
exception
)
{
errorFulfiller
->
reject
(
kj
::
mv
(
exception
));
}
// =======================================================================================
JsonRpc
::
ContentLengthTransport
::
ContentLengthTransport
(
kj
::
AsyncIoStream
&
stream
)
:
stream
(
stream
),
input
(
kj
::
newHttpInputStream
(
stream
,
staticHeaderTable
()))
{}
JsonRpc
::
ContentLengthTransport
::~
ContentLengthTransport
()
noexcept
(
false
)
{}
kj
::
Promise
<
void
>
JsonRpc
::
ContentLengthTransport
::
send
(
kj
::
StringPtr
text
)
{
auto
headers
=
kj
::
str
(
"Content-Length: "
,
text
.
size
(),
"
\r\n\r\n
"
);
parts
[
0
]
=
headers
.
asBytes
();
parts
[
1
]
=
text
.
asBytes
();
return
stream
.
write
(
parts
).
attach
(
kj
::
mv
(
headers
));
}
kj
::
Promise
<
kj
::
String
>
JsonRpc
::
ContentLengthTransport
::
receive
()
{
return
input
->
readMessage
()
.
then
([
this
](
kj
::
HttpInputStream
::
Message
&&
message
)
{
auto
promise
=
message
.
body
->
readAllText
();
return
promise
.
attach
(
kj
::
mv
(
message
.
body
));
});
}
}
// namespace capnp
c++/src/capnp/compat/json-rpc.capnp
0 → 100644
View file @
7260fe75
@0xd04299800d6725ba;
$import "/capnp/c++.capnp".namespace("capnp::json");
using Json = import "json.capnp";
struct RpcMessage {
jsonrpc @0 :Text;
# Must always be "2.0".
id @1 :Json.Value;
# Correlates a request to a response. Technically must be a string or number. Our implementation
# will always use a number for calls it initiates, and will reflect IDs of any type for calls
# it receives.
#
# May be omitted when caller doesn't care about the response. The implementation will omit `id`
# and return immediately when calling methods with the annotation `@notification` (defined in
# `json.capnp`). The `@notification` annotation only matters for outgoing calls; for incoming
# calls, it's the client's decision whether it wants to receive the response.
method @2 :Text;
# Method name. Only expected when `params` is sent.
union {
none @3 :Void $Json.name("!missing params, result, or error");
# Dummy default value of union, to detect when none of the fields below were received.
params @4 :Json.Value;
# Initiates a call.
result @5 :Json.Value;
# Completes a call.
error @6 :Error;
# Completes a call throwing an exception.
}
struct Error {
code @0 :Int32;
message @1 :Text;
data @2 :Json.Value;
}
}
c++/src/capnp/compat/json-rpc.h
0 → 100644
View file @
7260fe75
// Copyright (c) 2018 Kenton Varda and contributors
// Licensed under the MIT License:
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.
#pragma once
#include "json.h"
#include <kj/async-io.h>
#include <capnp/capability.h>
#include <kj/map.h>
namespace
kj
{
class
HttpInputStream
;
}
namespace
capnp
{
class
JsonRpc
:
private
kj
::
TaskSet
::
ErrorHandler
{
// An implementation of JSON-RPC 2.0: https://www.jsonrpc.org/specification
//
// This allows you to use Cap'n Proto interface declarations to implement JSON-RPC protocols.
// Of course, JSON-RPC does not support capabilities. So, the client and server each expose
// exactly one object to the other.
public
:
class
Transport
;
class
ContentLengthTransport
;
JsonRpc
(
Transport
&
transport
,
DynamicCapability
::
Client
interface
=
{});
KJ_DISALLOW_COPY
(
JsonRpc
);
DynamicCapability
::
Client
getPeer
(
InterfaceSchema
schema
);
template
<
typename
T
>
typename
T
::
Client
getPeer
()
{
return
getPeer
(
Schema
::
from
<
T
>
()).
template
castAs
<
T
>
();
}
kj
::
Promise
<
void
>
onError
()
{
return
errorPromise
.
addBranch
();
}
private
:
JsonCodec
codec
;
Transport
&
transport
;
DynamicCapability
::
Client
interface
;
kj
::
HashMap
<
kj
::
StringPtr
,
InterfaceSchema
::
Method
>
methodMap
;
uint
callCount
=
0
;
kj
::
Promise
<
void
>
writeQueue
=
kj
::
READY_NOW
;
kj
::
ForkedPromise
<
void
>
errorPromise
;
kj
::
Own
<
kj
::
PromiseFulfiller
<
void
>>
errorFulfiller
;
kj
::
Promise
<
void
>
readTask
;
struct
AwaitedResponse
{
CallContext
<
DynamicStruct
,
DynamicStruct
>
context
;
kj
::
Own
<
kj
::
PromiseFulfiller
<
void
>>
fulfiller
;
};
kj
::
HashMap
<
uint
,
AwaitedResponse
>
awaitedResponses
;
kj
::
TaskSet
tasks
;
class
CapabilityImpl
;
kj
::
Promise
<
void
>
queueWrite
(
kj
::
String
text
);
void
queueError
(
kj
::
Maybe
<
json
::
Value
::
Reader
>
id
,
int
code
,
kj
::
StringPtr
message
);
kj
::
Promise
<
void
>
readLoop
();
void
taskFailed
(
kj
::
Exception
&&
exception
)
override
;
JsonRpc
(
Transport
&
transport
,
DynamicCapability
::
Client
interface
,
kj
::
PromiseFulfillerPair
<
void
>
paf
);
};
class
JsonRpc
::
Transport
{
public
:
virtual
kj
::
Promise
<
void
>
send
(
kj
::
StringPtr
text
)
=
0
;
virtual
kj
::
Promise
<
kj
::
String
>
receive
()
=
0
;
};
class
JsonRpc
::
ContentLengthTransport
:
public
Transport
{
// The transport used by Visual Studio Code: Each message is composed like an HTTP message
// without the first line. That is, a list of headers, followed by a blank line, followed by the
// content whose length is determined by the content-length header.
public
:
explicit
ContentLengthTransport
(
kj
::
AsyncIoStream
&
stream
);
~
ContentLengthTransport
()
noexcept
(
false
);
KJ_DISALLOW_COPY
(
ContentLengthTransport
);
kj
::
Promise
<
void
>
send
(
kj
::
StringPtr
text
)
override
;
kj
::
Promise
<
kj
::
String
>
receive
()
override
;
private
:
kj
::
AsyncIoStream
&
stream
;
kj
::
Own
<
kj
::
HttpInputStream
>
input
;
kj
::
ArrayPtr
<
const
byte
>
parts
[
2
];
};
}
// namespace capnp
c++/src/capnp/compat/json.capnp
View file @
7260fe75
...
...
@@ -66,14 +66,14 @@ struct Value {
#
# myField @0 :Text $Json.name("my_field");
annotation name @0xfa5b1fd61c2e7c3d (field, enumerant, method, group, union)
:
Text;
annotation name @0xfa5b1fd61c2e7c3d (field, enumerant, method, group, union)
:
Text;
# Define an alternative name to use when encoding the given item in JSON. This can be used, for
# example, to use snake_case names where needed, even though Cap'n Proto uses strictly camelCase.
#
# (However, because JSON is derived from JavaScript, you *should* use camelCase names when
# defining JSON-based APIs. But, when supporting a pre-existing API you may not have a choice.)
annotation flatten @0x82d3e852af0336bf (field, group, union)
:
FlattenOptions;
annotation flatten @0x82d3e852af0336bf (field, group, union)
:
FlattenOptions;
# Specifies that an aggregate field should be flattened into its parent.
#
# In order to flatten a member of a union, the union (or, for an anonymous union, the parent
...
...
@@ -87,7 +87,7 @@ struct FlattenOptions {
# Optional: Adds the given prefix to flattened field names.
}
annotation discriminator @0xcfa794e8d19a0162 (struct, union)
:
DiscriminatorOptions;
annotation discriminator @0xcfa794e8d19a0162 (struct, union)
:
DiscriminatorOptions;
# Specifies that a union's variant will be decided not by which fields are present, but instead
# by a special discriminator field. The value of the discriminator field is a string naming which
# variant is active. This allows the members of the union to have the $jsonFlatten annotation, or
...
...
@@ -105,8 +105,11 @@ struct DiscriminatorOptions {
# It is an error to use `valueName` while also declaring some variants as $flatten.
}
annotation base64 @0xd7d879450a253e4b (field)
:
Void;
annotation base64 @0xd7d879450a253e4b (field)
:
Void;
# Place on a field of type `Data` to indicate that its JSON representation is a Base64 string.
annotation hex @0xf061e22f0ae5c7b5 (field)
:
Void;
annotation hex @0xf061e22f0ae5c7b5 (field)
:
Void;
# Place on a field of type `Data` to indicate that its JSON representation is a hex string.
annotation notification @0xa0a054dea32fd98c (method) :Void;
# Indicates that this method is a JSON-RPC "notification", meaning it expects no response.
c++/src/capnp/compat/json.h
View file @
7260fe75
...
...
@@ -303,6 +303,12 @@ void JsonCodec::encode(T&& value, JsonValue::Builder output) const {
encode
(
DynamicValue
::
Reader
(
ReaderFor
<
Base
>
(
kj
::
fwd
<
T
>
(
value
))),
Type
::
from
<
Base
>
(),
output
);
}
template
<>
inline
void
JsonCodec
::
encode
<
DynamicStruct
::
Reader
>
(
DynamicStruct
::
Reader
&&
value
,
JsonValue
::
Builder
output
)
const
{
encode
(
DynamicValue
::
Reader
(
value
),
value
.
getSchema
(),
output
);
}
template
<
typename
T
>
inline
Orphan
<
T
>
JsonCodec
::
decode
(
JsonValue
::
Reader
input
,
Orphanage
orphanage
)
const
{
return
decode
(
input
,
Type
::
from
<
T
>
(),
orphanage
).
template
releaseAs
<
T
>
();
...
...
c++/src/capnp/dynamic.h
View file @
7260fe75
...
...
@@ -584,6 +584,9 @@ public:
kj
::
Promise
<
void
>
tailCall
(
Request
<
SubParams
,
DynamicStruct
>&&
tailRequest
);
void
allowCancellation
();
StructSchema
getParamsType
()
const
{
return
paramType
;
}
StructSchema
getResultsType
()
const
{
return
resultType
;
}
private
:
CallContextHook
*
hook
;
StructSchema
paramType
;
...
...
c++/src/capnp/serialize-async.c++
View file @
7260fe75
...
...
@@ -71,9 +71,8 @@ kj::Promise<bool> AsyncMessageReader::read(kj::AsyncInputStream& inputStream,
return
false
;
}
else
if
(
n
<
sizeof
(
firstWord
))
{
// EOF in first word.
KJ_FAIL_REQUIRE
(
"Premature EOF."
)
{
return
false
;
}
kj
::
throwRecoverableException
(
KJ_EXCEPTION
(
DISCONNECTED
,
"Premature EOF."
));
return
false
;
}
return
readAfterFirstWord
(
inputStream
,
scratchSpace
).
then
([]()
{
return
true
;
});
...
...
@@ -153,7 +152,9 @@ kj::Promise<kj::Own<MessageReader>> readMessage(
auto
reader
=
kj
::
heap
<
AsyncMessageReader
>
(
options
);
auto
promise
=
reader
->
read
(
input
,
scratchSpace
);
return
promise
.
then
(
kj
::
mvCapture
(
reader
,
[](
kj
::
Own
<
MessageReader
>&&
reader
,
bool
success
)
{
KJ_REQUIRE
(
success
,
"Premature EOF."
)
{
break
;
}
if
(
!
success
)
{
kj
::
throwRecoverableException
(
KJ_EXCEPTION
(
DISCONNECTED
,
"Premature EOF."
));
}
return
kj
::
mv
(
reader
);
}));
}
...
...
c++/src/kj/compat/http-test.c++
View file @
7260fe75
...
...
@@ -1366,6 +1366,141 @@ KJ_TEST("HttpClient <-> HttpServer") {
// -----------------------------------------------------------------------------
KJ_TEST
(
"HttpInputStream requests"
)
{
kj
::
EventLoop
eventLoop
;
kj
::
WaitScope
waitScope
(
eventLoop
);
kj
::
HttpHeaderTable
table
;
auto
pipe
=
kj
::
newOneWayPipe
();
auto
input
=
newHttpInputStream
(
*
pipe
.
in
,
table
);
kj
::
Promise
<
void
>
writeQueue
=
kj
::
READY_NOW
;
for
(
auto
&
testCase
:
requestTestCases
())
{
writeQueue
=
writeQueue
.
then
([
&
]()
{
return
pipe
.
out
->
write
(
testCase
.
raw
.
begin
(),
testCase
.
raw
.
size
());
});
}
writeQueue
=
writeQueue
.
then
([
&
]()
{
pipe
.
out
=
nullptr
;
});
for
(
auto
&
testCase
:
requestTestCases
())
{
KJ_CONTEXT
(
testCase
.
raw
);
KJ_ASSERT
(
input
->
awaitNextMessage
().
wait
(
waitScope
));
auto
req
=
input
->
readRequest
().
wait
(
waitScope
);
KJ_EXPECT
(
req
.
method
==
testCase
.
method
);
KJ_EXPECT
(
req
.
url
==
testCase
.
path
);
for
(
auto
&
header
:
testCase
.
requestHeaders
)
{
KJ_EXPECT
(
KJ_ASSERT_NONNULL
(
req
.
headers
.
get
(
header
.
id
))
==
header
.
value
);
}
auto
body
=
req
.
body
->
readAllText
().
wait
(
waitScope
);
KJ_EXPECT
(
body
==
kj
::
strArray
(
testCase
.
requestBodyParts
,
""
));
}
writeQueue
.
wait
(
waitScope
);
KJ_EXPECT
(
!
input
->
awaitNextMessage
().
wait
(
waitScope
));
}
KJ_TEST
(
"HttpInputStream responses"
)
{
kj
::
EventLoop
eventLoop
;
kj
::
WaitScope
waitScope
(
eventLoop
);
kj
::
HttpHeaderTable
table
;
auto
pipe
=
kj
::
newOneWayPipe
();
auto
input
=
newHttpInputStream
(
*
pipe
.
in
,
table
);
kj
::
Promise
<
void
>
writeQueue
=
kj
::
READY_NOW
;
for
(
auto
&
testCase
:
responseTestCases
())
{
if
(
testCase
.
side
==
CLIENT_ONLY
)
continue
;
// skip Connection: close case.
writeQueue
=
writeQueue
.
then
([
&
]()
{
return
pipe
.
out
->
write
(
testCase
.
raw
.
begin
(),
testCase
.
raw
.
size
());
});
}
writeQueue
=
writeQueue
.
then
([
&
]()
{
pipe
.
out
=
nullptr
;
});
for
(
auto
&
testCase
:
responseTestCases
())
{
if
(
testCase
.
side
==
CLIENT_ONLY
)
continue
;
// skip Connection: close case.
KJ_CONTEXT
(
testCase
.
raw
);
KJ_ASSERT
(
input
->
awaitNextMessage
().
wait
(
waitScope
));
auto
resp
=
input
->
readResponse
(
testCase
.
method
).
wait
(
waitScope
);
KJ_EXPECT
(
resp
.
statusCode
==
testCase
.
statusCode
);
KJ_EXPECT
(
resp
.
statusText
==
testCase
.
statusText
);
for
(
auto
&
header
:
testCase
.
responseHeaders
)
{
KJ_EXPECT
(
KJ_ASSERT_NONNULL
(
resp
.
headers
.
get
(
header
.
id
))
==
header
.
value
);
}
auto
body
=
resp
.
body
->
readAllText
().
wait
(
waitScope
);
KJ_EXPECT
(
body
==
kj
::
strArray
(
testCase
.
responseBodyParts
,
""
));
}
writeQueue
.
wait
(
waitScope
);
KJ_EXPECT
(
!
input
->
awaitNextMessage
().
wait
(
waitScope
));
}
KJ_TEST
(
"HttpInputStream bare messages"
)
{
kj
::
EventLoop
eventLoop
;
kj
::
WaitScope
waitScope
(
eventLoop
);
kj
::
HttpHeaderTable
table
;
auto
pipe
=
kj
::
newOneWayPipe
();
auto
input
=
newHttpInputStream
(
*
pipe
.
in
,
table
);
kj
::
StringPtr
messages
=
"Content-Length: 6
\r\n
"
"
\r\n
"
"foobar"
"Content-Length: 11
\r\n
"
"Content-Type: some/type
\r\n
"
"
\r\n
"
"bazquxcorge"
"Transfer-Encoding: chunked
\r\n
"
"
\r\n
"
"6
\r\n
"
"grault
\r\n
"
"b
\r\n
"
"garplywaldo
\r\n
"
"0
\r\n
"
"
\r\n
"
_kj
;
kj
::
Promise
<
void
>
writeTask
=
pipe
.
out
->
write
(
messages
.
begin
(),
messages
.
size
())
.
then
([
&
]()
{
pipe
.
out
=
nullptr
;
});
{
KJ_ASSERT
(
input
->
awaitNextMessage
().
wait
(
waitScope
));
auto
message
=
input
->
readMessage
().
wait
(
waitScope
);
KJ_EXPECT
(
KJ_ASSERT_NONNULL
(
message
.
headers
.
get
(
HttpHeaderId
::
CONTENT_LENGTH
))
==
"6"
);
KJ_EXPECT
(
message
.
body
->
readAllText
().
wait
(
waitScope
)
==
"foobar"
);
}
{
KJ_ASSERT
(
input
->
awaitNextMessage
().
wait
(
waitScope
));
auto
message
=
input
->
readMessage
().
wait
(
waitScope
);
KJ_EXPECT
(
KJ_ASSERT_NONNULL
(
message
.
headers
.
get
(
HttpHeaderId
::
CONTENT_LENGTH
))
==
"11"
);
KJ_EXPECT
(
KJ_ASSERT_NONNULL
(
message
.
headers
.
get
(
HttpHeaderId
::
CONTENT_TYPE
))
==
"some/type"
);
KJ_EXPECT
(
message
.
body
->
readAllText
().
wait
(
waitScope
)
==
"bazquxcorge"
);
}
{
KJ_ASSERT
(
input
->
awaitNextMessage
().
wait
(
waitScope
));
auto
message
=
input
->
readMessage
().
wait
(
waitScope
);
KJ_EXPECT
(
KJ_ASSERT_NONNULL
(
message
.
headers
.
get
(
HttpHeaderId
::
TRANSFER_ENCODING
))
==
"chunked"
);
KJ_EXPECT
(
message
.
body
->
readAllText
().
wait
(
waitScope
)
==
"graultgarplywaldo"
);
}
writeTask
.
wait
(
waitScope
);
KJ_EXPECT
(
!
input
->
awaitNextMessage
().
wait
(
waitScope
));
}
// -----------------------------------------------------------------------------
KJ_TEST
(
"WebSocket core protocol"
)
{
kj
::
EventLoop
eventLoop
;
kj
::
WaitScope
waitScope
(
eventLoop
);
...
...
c++/src/kj/compat/http.c++
View file @
7260fe75
...
...
@@ -903,6 +903,14 @@ kj::Maybe<HttpHeaders::Response> HttpHeaders::tryParseResponse(kj::ArrayPtr<char
return
response
;
}
bool
HttpHeaders
::
tryParse
(
kj
::
ArrayPtr
<
char
>
content
)
{
char
*
end
=
trimHeaderEnding
(
content
);
if
(
end
==
nullptr
)
return
false
;
char
*
ptr
=
content
.
begin
();
return
parseHeaders
(
ptr
,
end
);
}
bool
HttpHeaders
::
parseHeaders
(
char
*
ptr
,
char
*
end
)
{
while
(
*
ptr
!=
'\0'
)
{
KJ_IF_MAYBE
(
name
,
consumeHeaderName
(
ptr
))
{
...
...
@@ -988,9 +996,9 @@ static constexpr size_t MIN_BUFFER = 4096;
static
constexpr
size_t
MAX_BUFFER
=
65536
;
static
constexpr
size_t
MAX_CHUNK_HEADER_SIZE
=
32
;
class
HttpInputStream
{
class
HttpInputStream
Impl
final
:
public
HttpInputStream
{
public
:
explicit
HttpInputStream
(
AsyncIo
Stream
&
inner
,
HttpHeaderTable
&
table
)
explicit
HttpInputStream
Impl
(
AsyncInput
Stream
&
inner
,
HttpHeaderTable
&
table
)
:
inner
(
inner
),
headerBuffer
(
kj
::
heapArray
<
char
>
(
MIN_BUFFER
)),
headers
(
table
)
{
}
...
...
@@ -998,6 +1006,41 @@ public:
return
!
broken
&&
pendingMessageCount
==
0
;
}
// ---------------------------------------------------------------------------
// public interface
kj
::
Promise
<
Request
>
readRequest
()
override
{
return
readRequestHeaders
()
.
then
([
this
](
kj
::
Maybe
<
HttpHeaders
::
Request
>&&
maybeRequest
)
->
HttpInputStream
::
Request
{
auto
request
=
KJ_REQUIRE_NONNULL
(
maybeRequest
,
"bad request"
);
auto
body
=
getEntityBody
(
HttpInputStreamImpl
::
REQUEST
,
request
.
method
,
0
,
headers
);
return
{
request
.
method
,
request
.
url
,
headers
,
kj
::
mv
(
body
)
};
});
}
kj
::
Promise
<
Response
>
readResponse
(
HttpMethod
requestMethod
)
override
{
return
readResponseHeaders
()
.
then
([
this
,
requestMethod
](
kj
::
Maybe
<
HttpHeaders
::
Response
>&&
maybeResponse
)
->
HttpInputStream
::
Response
{
auto
response
=
KJ_REQUIRE_NONNULL
(
maybeResponse
,
"bad response"
);
auto
body
=
getEntityBody
(
HttpInputStreamImpl
::
RESPONSE
,
requestMethod
,
0
,
headers
);
return
{
response
.
statusCode
,
response
.
statusText
,
headers
,
kj
::
mv
(
body
)
};
});
}
kj
::
Promise
<
Message
>
readMessage
()
override
{
return
readMessageHeaders
()
.
then
([
this
](
kj
::
ArrayPtr
<
char
>
text
)
->
HttpInputStream
::
Message
{
headers
.
clear
();
KJ_REQUIRE
(
headers
.
tryParse
(
text
),
"bad message"
);
auto
body
=
getEntityBody
(
HttpInputStreamImpl
::
RESPONSE
,
HttpMethod
::
GET
,
0
,
headers
);
return
{
headers
,
kj
::
mv
(
body
)
};
});
}
// ---------------------------------------------------------------------------
// Stream locking: While an entity-body is being read, the body stream "locks" the underlying
// HTTP stream. Once the entity-body is complete, we can read the next pipelined message.
...
...
@@ -1022,7 +1065,7 @@ public:
// ---------------------------------------------------------------------------
kj
::
Promise
<
bool
>
awaitNextMessage
()
{
kj
::
Promise
<
bool
>
awaitNextMessage
()
override
{
// Waits until more data is available, but doesn't consume it. Returns false on EOF.
//
// Used on the server after a request is handled, to check for pipelined requests.
...
...
@@ -1172,7 +1215,7 @@ public:
}
private
:
AsyncI
o
Stream
&
inner
;
AsyncI
nput
Stream
&
inner
;
kj
::
Array
<
char
>
headerBuffer
;
size_t
messageHeaderEnd
=
0
;
...
...
@@ -1367,7 +1410,7 @@ private:
class
HttpEntityBodyReader
:
public
kj
::
AsyncInputStream
{
public
:
HttpEntityBodyReader
(
HttpInputStream
&
inner
)
:
inner
(
inner
)
{}
HttpEntityBodyReader
(
HttpInputStream
Impl
&
inner
)
:
inner
(
inner
)
{}
~
HttpEntityBodyReader
()
noexcept
(
false
)
{
if
(
!
finished
)
{
inner
.
abortRead
();
...
...
@@ -1375,7 +1418,7 @@ public:
}
protected
:
HttpInputStream
&
inner
;
HttpInputStream
Impl
&
inner
;
void
doneReading
()
{
KJ_REQUIRE
(
!
finished
);
...
...
@@ -1394,7 +1437,7 @@ class HttpNullEntityReader final: public HttpEntityBodyReader {
// may indicate non-zero in the special case of a response to a HEAD request.
public
:
HttpNullEntityReader
(
HttpInputStream
&
inner
,
kj
::
Maybe
<
uint64_t
>
length
)
HttpNullEntityReader
(
HttpInputStream
Impl
&
inner
,
kj
::
Maybe
<
uint64_t
>
length
)
:
HttpEntityBodyReader
(
inner
),
length
(
length
)
{
// `length` is what to return from tryGetLength(). For a response to a HEAD request, this may
// be non-zero.
...
...
@@ -1417,7 +1460,7 @@ class HttpConnectionCloseEntityReader final: public HttpEntityBodyReader {
// Stream which reads until EOF.
public
:
HttpConnectionCloseEntityReader
(
HttpInputStream
&
inner
)
HttpConnectionCloseEntityReader
(
HttpInputStream
Impl
&
inner
)
:
HttpEntityBodyReader
(
inner
)
{}
Promise
<
size_t
>
tryRead
(
void
*
buffer
,
size_t
minBytes
,
size_t
maxBytes
)
override
{
...
...
@@ -1437,7 +1480,7 @@ class HttpFixedLengthEntityReader final: public HttpEntityBodyReader {
// Stream which reads only up to a fixed length from the underlying stream, then emulates EOF.
public
:
HttpFixedLengthEntityReader
(
HttpInputStream
&
inner
,
size_t
length
)
HttpFixedLengthEntityReader
(
HttpInputStream
Impl
&
inner
,
size_t
length
)
:
HttpEntityBodyReader
(
inner
),
length
(
length
)
{
if
(
length
==
0
)
doneReading
();
}
...
...
@@ -1470,7 +1513,7 @@ class HttpChunkedEntityReader final: public HttpEntityBodyReader {
// Stream which reads a Transfer-Encoding: Chunked stream.
public
:
HttpChunkedEntityReader
(
HttpInputStream
&
inner
)
HttpChunkedEntityReader
(
HttpInputStream
Impl
&
inner
)
:
HttpEntityBodyReader
(
inner
)
{}
Promise
<
size_t
>
tryRead
(
void
*
buffer
,
size_t
minBytes
,
size_t
maxBytes
)
override
{
...
...
@@ -1551,7 +1594,7 @@ static_assert(!fastCaseCmp<'n','O','o','B','1'>("FooB1"), "");
static_assert
(
!
fastCaseCmp
<
'f'
,
'O'
,
'o'
,
'B'
>
(
"FooB1"
),
""
);
static_assert
(
!
fastCaseCmp
<
'f'
,
'O'
,
'o'
,
'B'
,
'1'
,
'a'
>
(
"FooB1"
),
""
);
kj
::
Own
<
kj
::
AsyncInputStream
>
HttpInputStream
::
getEntityBody
(
kj
::
Own
<
kj
::
AsyncInputStream
>
HttpInputStream
Impl
::
getEntityBody
(
RequestOrResponse
type
,
HttpMethod
method
,
uint
statusCode
,
const
kj
::
HttpHeaders
&
headers
)
{
if
(
type
==
RESPONSE
)
{
...
...
@@ -1599,8 +1642,16 @@ kj::Own<kj::AsyncInputStream> HttpInputStream::getEntityBody(
return
kj
::
heap
<
HttpNullEntityReader
>
(
*
this
,
uint64_t
(
0
));
}
}
// namespace
kj
::
Own
<
HttpInputStream
>
newHttpInputStream
(
kj
::
AsyncInputStream
&
input
,
HttpHeaderTable
&
table
)
{
return
kj
::
heap
<
HttpInputStreamImpl
>
(
input
,
table
);
}
// =======================================================================================
namespace
{
class
HttpOutputStream
{
public
:
HttpOutputStream
(
AsyncOutputStream
&
inner
)
:
inner
(
inner
)
{}
...
...
@@ -2397,7 +2448,7 @@ private:
};
kj
::
Own
<
WebSocket
>
upgradeToWebSocket
(
kj
::
Own
<
kj
::
AsyncIoStream
>
stream
,
HttpInputStream
&
httpInput
,
HttpOutputStream
&
httpOutput
,
kj
::
Own
<
kj
::
AsyncIoStream
>
stream
,
HttpInputStream
Impl
&
httpInput
,
HttpOutputStream
&
httpOutput
,
kj
::
Maybe
<
EntropySource
&>
maskKeyGenerator
)
{
// Create a WebSocket upgraded from an HTTP stream.
auto
releasedBuffer
=
httpInput
.
releaseBuffer
();
...
...
@@ -3064,7 +3115,7 @@ public:
r
->
statusCode
,
r
->
statusText
,
&
headers
,
httpInput
.
getEntityBody
(
HttpInputStream
::
RESPONSE
,
method
,
r
->
statusCode
,
headers
)
httpInput
.
getEntityBody
(
HttpInputStream
Impl
::
RESPONSE
,
method
,
r
->
statusCode
,
headers
)
};
if
(
fastCaseCmp
<
'c'
,
'l'
,
'o'
,
's'
,
'e'
>
(
...
...
@@ -3156,7 +3207,7 @@ public:
r
->
statusCode
,
r
->
statusText
,
&
headers
,
httpInput
.
getEntityBody
(
HttpInputStream
::
RESPONSE
,
HttpMethod
::
GET
,
r
->
statusCode
,
httpInput
.
getEntityBody
(
HttpInputStream
Impl
::
RESPONSE
,
HttpMethod
::
GET
,
r
->
statusCode
,
headers
)
};
if
(
fastCaseCmp
<
'c'
,
'l'
,
'o'
,
's'
,
'e'
>
(
...
...
@@ -3178,7 +3229,7 @@ public:
}
private
:
HttpInputStream
httpInput
;
HttpInputStream
Impl
httpInput
;
HttpOutputStream
httpOutput
;
kj
::
Own
<
AsyncIoStream
>
ownStream
;
HttpClientSettings
settings
;
...
...
@@ -4159,7 +4210,7 @@ public:
currentMethod
=
req
->
method
;
auto
body
=
httpInput
.
getEntityBody
(
HttpInputStream
::
REQUEST
,
req
->
method
,
0
,
headers
);
HttpInputStream
Impl
::
REQUEST
,
req
->
method
,
0
,
headers
);
// TODO(perf): If the client disconnects, should we cancel the response? Probably, to
// prevent permanent deadlock. It's slightly weird in that arguably the client should
...
...
@@ -4312,7 +4363,7 @@ private:
HttpServer
&
server
;
kj
::
AsyncIoStream
&
stream
;
HttpService
&
service
;
HttpInputStream
httpInput
;
HttpInputStream
Impl
httpInput
;
HttpOutputStream
httpOutput
;
kj
::
Maybe
<
HttpMethod
>
currentMethod
;
bool
timedOut
=
false
;
...
...
c++/src/kj/compat/http.h
View file @
7260fe75
...
...
@@ -332,6 +332,9 @@ public:
// to split it into a bunch of shorter strings. The caller must keep `content` valid until the
// `HttpHeaders` is destroyed, or pass it to `takeOwnership()`.
bool
tryParse
(
kj
::
ArrayPtr
<
char
>
content
);
// Like tryParseRequest()/tryParseResponse(), but don't expect any request/response line.
kj
::
String
serializeRequest
(
HttpMethod
method
,
kj
::
StringPtr
url
,
kj
::
ArrayPtr
<
const
kj
::
StringPtr
>
connectionHeaders
=
nullptr
)
const
;
kj
::
String
serializeResponse
(
uint
statusCode
,
kj
::
StringPtr
statusText
,
...
...
@@ -396,6 +399,58 @@ private:
// also add direct accessors for those headers.
};
class
HttpInputStream
{
// Low-level interface to receive HTTP-formatted messages (headers followed by body) from an
// input stream, without a paired output stream.
//
// Most applications will not use this. Regular HTTP clients and servers don't need this. This
// is mainly useful for apps implementing various protocols that look like HTTP but aren't
// really.
public
:
struct
Request
{
HttpMethod
method
;
kj
::
StringPtr
url
;
const
HttpHeaders
&
headers
;
kj
::
Own
<
kj
::
AsyncInputStream
>
body
;
};
virtual
kj
::
Promise
<
Request
>
readRequest
()
=
0
;
// Reads one HTTP request from the input stream.
//
// The returned struct contains pointers directly into a buffer that is invalidated on the next
// message read.
struct
Response
{
uint
statusCode
;
kj
::
StringPtr
statusText
;
const
HttpHeaders
&
headers
;
kj
::
Own
<
kj
::
AsyncInputStream
>
body
;
};
virtual
kj
::
Promise
<
Response
>
readResponse
(
HttpMethod
requestMethod
)
=
0
;
// Reads one HTTP response from the input stream.
//
// You must provide the request method because responses to HEAD requests require special
// treatment.
//
// The returned struct contains pointers directly into a buffer that is invalidated on the next
// message read.
struct
Message
{
const
HttpHeaders
&
headers
;
kj
::
Own
<
kj
::
AsyncInputStream
>
body
;
};
virtual
kj
::
Promise
<
Message
>
readMessage
()
=
0
;
// Reads an HTTP header set followed by a body, with no request or response line. This is not
// useful for HTTP but may be useful for other protocols that make the unfortunate choice to
// mimic HTTP message format, such as Visual Studio Code's JSON-RPC transport.
//
// The returned struct contains pointers directly into a buffer that is invalidated on the next
// message read.
virtual
kj
::
Promise
<
bool
>
awaitNextMessage
()
=
0
;
// Waits until more data is available, but doesn't consume it. Returns false on EOF.
};
class
EntropySource
{
// Interface for an object that generates entropy. Typically, cryptographically-random entropy
// is expected.
...
...
@@ -641,6 +696,16 @@ kj::Own<HttpClient> newHttpClient(HttpService& service);
kj
::
Own
<
HttpService
>
newHttpService
(
HttpClient
&
client
);
// Adapts an HttpClient to an HttpService and vice versa.
kj
::
Own
<
HttpInputStream
>
newHttpInputStream
(
kj
::
AsyncInputStream
&
input
,
HttpHeaderTable
&
headerTable
);
// Create an HttpInputStream on top of the given stream. Normally applications would not call this
// directly, but it can be useful for implementing protocols that aren't quite HTTP but use similar
// message delimiting.
//
// The HttpInputStream implementation does read-ahead buffering on `input`. Therefore, when the
// HttpInputStream is destroyed, some data read from `input` may be lost, so it's not possible to
// continue reading from `input` in a reliable way.
kj
::
Own
<
WebSocket
>
newWebSocket
(
kj
::
Own
<
kj
::
AsyncIoStream
>
stream
,
kj
::
Maybe
<
EntropySource
&>
maskEntropySource
);
// Create a new WebSocket on top of the given stream. It is assumed that the HTTP -> WebSocket
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment