Skip to content

Commit

Permalink
Update to the oatpp 0.19.12 API.
Browse files Browse the repository at this point in the history
  • Loading branch information
@lganzzzo
lganzzzo committed Jan 11, 2020
1 parent be56875 commit a71fa8d
Show file tree
Hide file tree
Showing 12 changed files with 92 additions and 72 deletions.
2 changes: 1 addition & 1 deletion CMakeLists.txt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@ cmake_minimum_required(VERSION 3.1 FATAL_ERROR)
## use these variables to configure module installation

set(OATPP_THIS_MODULE_NAME oatpp-websocket) ## name of the module (also name of folders in installation dirs)
set(OATPP_THIS_MODULE_VERSION "0.19.11") ## version of the module (also sufix of folders in installation dirs)
set(OATPP_THIS_MODULE_VERSION "0.19.12") ## version of the module (also sufix of folders in installation dirs)
set(OATPP_THIS_MODULE_LIBRARIES oatpp-websocket) ## list of libraries to find when find_package is called
set(OATPP_THIS_MODULE_TARGETS oatpp-websocket) ## list of targets to install
set(OATPP_THIS_MODULE_DIRECTORIES oatpp-websocket) ## list of directories to install
Expand Down
5 changes: 4 additions & 1 deletion src/oatpp-websocket/AsyncConnectionHandler.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,10 @@ void AsyncConnectionHandler::handleConnection(const std::shared_ptr<IOStream>& c
}

};


connection->setInputStreamIOMode(oatpp::data::stream::IOMode::ASYNCHRONOUS);
connection->setOutputStreamIOMode(oatpp::data::stream::IOMode::ASYNCHRONOUS);

m_executor->execute<SocketCoroutine>(connection, params, m_listener);

}
Expand Down
44 changes: 24 additions & 20 deletions src/oatpp-websocket/AsyncWebSocket.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readFrameHeaderAsync(const std::s
v_word16 m_messageLen2;
v_word32 m_messageLen3 [2];
private:
oatpp::data::stream::AsyncInlineReadData m_inlineData;
oatpp::data::buffer::InlineReadData m_inlineData;
public:

ReadFrameCoroutine(const std::shared_ptr<oatpp::data::stream::IOStream> connection,
Expand All @@ -91,7 +91,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readFrameHeaderAsync(const std::s
}

Action act() override {
return oatpp::data::stream::readExactSizeDataAsyncInline(m_connection.get(), m_inlineData, yieldTo(&ReadFrameCoroutine::onBbRead));
return m_connection->readExactSizeDataAsyncInline(m_inlineData, yieldTo(&ReadFrameCoroutine::onBbRead));
}

Action onBbRead() {
Expand All @@ -118,7 +118,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readFrameHeaderAsync(const std::s
}

Action readLen() {
return oatpp::data::stream::readExactSizeDataAsyncInline(m_connection.get(), m_inlineData, yieldTo(&ReadFrameCoroutine::onLenRead));
return m_connection->readExactSizeDataAsyncInline(m_inlineData, yieldTo(&ReadFrameCoroutine::onLenRead));
}

Action onLenRead() {
Expand All @@ -139,7 +139,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readFrameHeaderAsync(const std::s
}

Action readMask() {
return oatpp::data::stream::readExactSizeDataAsyncInline(m_connection.get(), m_inlineData, finish());
return m_connection->readExactSizeDataAsyncInline(m_inlineData, finish());
}

};
Expand All @@ -162,7 +162,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::writeFrameHeaderAsync(const std::
v_word32 m_messageLen3 [2];
v_word8 m_messageLengthScenario;
private:
oatpp::data::stream::AsyncInlineWriteData m_inlineData;
oatpp::data::buffer::InlineWriteData m_inlineData;
public:

WriteFrameCoroutine(const std::shared_ptr<oatpp::data::stream::IOStream> connection,
Expand All @@ -177,7 +177,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::writeFrameHeaderAsync(const std::
}

Action act() override {
return oatpp::data::stream::writeExactSizeDataAsyncInline(m_connection.get(), m_inlineData, yieldTo(&WriteFrameCoroutine::onBbWritten));
return m_connection->writeExactSizeDataAsyncInline(m_inlineData, yieldTo(&WriteFrameCoroutine::onBbWritten));
}

Action onBbWritten() {
Expand All @@ -195,7 +195,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::writeFrameHeaderAsync(const std::
}

Action writeMessageLen() {
return oatpp::data::stream::writeExactSizeDataAsyncInline(m_connection.get(), m_inlineData, yieldTo(&WriteFrameCoroutine::onLenWritten));
return m_connection->writeExactSizeDataAsyncInline(m_inlineData, yieldTo(&WriteFrameCoroutine::onLenWritten));
}

Action onLenWritten() {
Expand All @@ -207,7 +207,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::writeFrameHeaderAsync(const std::
}

Action writeMask() {
return oatpp::data::stream::writeExactSizeDataAsyncInline(m_connection.get(), m_inlineData, finish());
return m_connection->writeExactSizeDataAsyncInline(m_inlineData, finish());
}

};
Expand All @@ -229,9 +229,9 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readPayloadAsync(const std::share
std::shared_ptr<Listener> m_listener;
private:
p_char8 m_buffer;
oatpp::data::v_io_size m_progress;
oatpp::v_io_size m_progress;
private:
oatpp::data::stream::AsyncInlineReadData m_inlineData;
oatpp::data::buffer::InlineReadData m_inlineData;
public:
ReadPayloadCoroutine(const std::shared_ptr<AsyncWebSocket>& socket,
const std::shared_ptr<oatpp::data::stream::IOStream>& connection,
Expand All @@ -245,7 +245,11 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readPayloadAsync(const std::share
, m_listener(listener)
, m_buffer(new v_char8[m_socket->m_config.readBufferSize])
, m_progress(0)
{}
{
if(m_shortMessageStream && m_frameHeader->payloadLength > 125) {
throw std::runtime_error("[oatpp::web::protocol::websocket::AsyncWebSocket::readPayloadAsync()]: Invalid payloadLength. See RFC-6455, section-5.5.");
}
}

~ReadPayloadCoroutine() {
delete [] m_buffer;
Expand All @@ -255,7 +259,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readPayloadAsync(const std::share

if(m_progress < m_frameHeader->payloadLength) {

oatpp::data::v_io_size desiredSize = m_socket->m_config.readBufferSize;
oatpp::v_io_size desiredSize = m_socket->m_config.readBufferSize;
if(desiredSize > m_frameHeader->payloadLength - m_progress) {
desiredSize = m_frameHeader->payloadLength - m_progress;
}
Expand All @@ -273,7 +277,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readPayloadAsync(const std::share
}

Action readData() {
return oatpp::data::stream::readSomeDataAsyncInline(m_connection.get(), m_inlineData, yieldTo(&ReadPayloadCoroutine::onDataRead));
return m_connection->readSomeDataAsyncInline(m_inlineData, yieldTo(&ReadPayloadCoroutine::onDataRead));
}

Action onDataRead() {
Expand All @@ -283,7 +287,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readPayloadAsync(const std::share
if(readResult > 0) {

if(m_frameHeader->hasMask) {
std::unique_ptr<v_char8> decoded(new v_char8[readResult]);
std::unique_ptr<v_char8> decoded(new v_char8[readResult]);
for(v_int32 i = 0; i < readResult; i ++) {
decoded.get()[i] = m_buffer[i] ^ m_frameHeader->mask[(i + m_progress) % 4];
}
Expand All @@ -292,7 +296,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readPayloadAsync(const std::share

if(m_shortMessageStream) {
/* this is RAM stream. Non-blocking call */
m_shortMessageStream->write(decoded.get(), readResult);
m_shortMessageStream->writeSimple(decoded.get(), readResult);
} else if(m_listener) {
return m_listener->readMessage(m_socket, m_frameHeader->opcode, decoded.get(), readResult).next(yieldTo(&ReadPayloadCoroutine::act));
}
Expand All @@ -303,7 +307,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::readPayloadAsync(const std::share

if(m_shortMessageStream) {
/* this is RAM stream. Non-blocking call */
m_shortMessageStream->write(m_buffer, readResult);
m_shortMessageStream->writeSimple(m_buffer, readResult);
} else if(m_listener) {
return m_listener->readMessage(m_socket, m_frameHeader->opcode, m_buffer, readResult).next(yieldTo(&ReadPayloadCoroutine::act));
}
Expand Down Expand Up @@ -487,7 +491,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::sendOneFrameAsync(bool fin, v_wor
private:
p_char8 m_encoded = nullptr;
private:
oatpp::data::stream::AsyncInlineWriteData m_inlineData;
oatpp::data::buffer::InlineWriteData m_inlineData;
public:
SendFrameCoroutine(const std::shared_ptr<AsyncWebSocket>& socket,
bool fin, v_word8 opcode, const oatpp::String& message)
Expand Down Expand Up @@ -526,7 +530,7 @@ oatpp::async::CoroutineStarter AsyncWebSocket::sendOneFrameAsync(bool fin, v_wor
}

Action writeMessage() {
return oatpp::data::stream::writeExactSizeDataAsyncInline(m_socket->m_connection.get(), m_inlineData, finish());
return m_socket->m_connection->writeExactSizeDataAsyncInline(m_inlineData, finish());
}

};
Expand All @@ -540,9 +544,9 @@ oatpp::async::CoroutineStarter AsyncWebSocket::sendCloseAsync(v_word16 code, con
code = htons(code);

oatpp::data::stream::ChunkedBuffer buffer;
buffer.write(&code, 2);
buffer.writeSimple(&code, 2);
if(message) {
buffer.write(message->getData(), message->getSize());
buffer.writeSimple(message->getData(), message->getSize());
}

return sendOneFrameAsync(true, Frame::OPCODE_CLOSE, buffer.toString());
Expand Down
4 changes: 2 additions & 2 deletions src/oatpp-websocket/AsyncWebSocket.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -110,11 +110,11 @@ class AsyncWebSocket : public oatpp::base::Countable, public std::enable_shared_
* @param socket - &id:oatpp::websocket::AsyncWebSocket;.
* @param opcode - &id:oatpp::websocket::Frame::OPCODE_TEXT; or &id:oatpp::websocket::Frame::OPCODE_BINARY;.
* @param data - pointer to received data.
* @param size - data size. &id:oatpp::data::v_io_size;.
* @param size - data size. &id:oatpp::v_io_size;.
* @return - &id:oatpp::async::CoroutineStarter;. <br>
* *To ignore this event return nullptr.*
*/
virtual CoroutineStarter readMessage(const std::shared_ptr<AsyncWebSocket>& socket, v_word8 opcode, p_char8 data, data::v_io_size size) = 0;
virtual CoroutineStarter readMessage(const std::shared_ptr<AsyncWebSocket>& socket, v_word8 opcode, p_char8 data, oatpp::v_io_size size) = 0;

};

Expand Down
3 changes: 3 additions & 0 deletions src/oatpp-websocket/ConnectionHandler.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,9 @@ void ConnectionHandler::handleConnection(const std::shared_ptr<IOStream>& connec
}

};

connection->setInputStreamIOMode(oatpp::data::stream::IOMode::BLOCKING);
connection->setOutputStreamIOMode(oatpp::data::stream::IOMode::BLOCKING);

/* Create working thread */
std::thread thread(&Task::run, Task(connection, params, m_listener));
Expand Down
14 changes: 7 additions & 7 deletions src/oatpp-websocket/SHA1.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -227,15 +227,15 @@ SHA1::SHA1() {


void SHA1::update(const oatpp::String &s) {
data::v_io_size progress = 0;
oatpp::v_io_size progress = 0;
while (true) {

auto readSize = BLOCK_BYTES - buffer.getSize();
if(readSize > s->getSize() - progress) {
readSize = s->getSize() - progress;
}

buffer.write(&s->getData()[progress], readSize);
buffer.writeSimple(&s->getData()[progress], readSize);
progress += readSize;

if (buffer.getSize() != BLOCK_BYTES) {
Expand All @@ -256,8 +256,8 @@ void SHA1::update(std::istream &is) {

char sbuf[BLOCK_BYTES];

is.read(sbuf, (int)(BLOCK_BYTES - buffer.getSize()));
buffer.write(sbuf, (std::size_t)is.gcount());
is.read(sbuf, (int)(BLOCK_BYTES - buffer.getSize()));
buffer.writeSimple(sbuf, (std::size_t)is.gcount());

if (buffer.getSize() != BLOCK_BYTES) {
return;
Expand All @@ -281,10 +281,10 @@ oatpp::String SHA1::finalBinary() {
uint64_t total_bits = (transforms * BLOCK_BYTES + buffer.getSize()) * 8;

/* Padding */
buffer.writeChar(0x80);
buffer.writeCharSimple(0x80);
size_t orig_size = (size_t)buffer.getSize();
while (buffer.getSize() < BLOCK_BYTES) {
buffer.writeChar(0x00);
buffer.writeCharSimple(0x00);
}

uint32_t block[BLOCK_INTS];
Expand All @@ -306,7 +306,7 @@ oatpp::String SHA1::finalBinary() {
oatpp::data::stream::ChunkedBuffer resultStream;
for (size_t i = 0; i < sizeof(digest) / sizeof(digest[0]); i++) {
uint32_t b = htonl(digest[i]);
resultStream.write(&b, 4);
resultStream.writeSimple(&b, 4);
}

/* Reset for next run */
Expand Down
Loading

0 comments on commit a71fa8d

Please sign in to comment.