Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
84d78035d1 | ||
|
|
719487c5fb | ||
|
|
d9f3863d23 | ||
|
|
c2fe604e91 | ||
|
|
799d8f0334 |
+30
-27
@@ -52,7 +52,8 @@ void StreamAPI::writeStream()
|
||||
do {
|
||||
// Send every packet we can
|
||||
len = getFromRadio(txBuf + HEADER_LEN);
|
||||
emitTxBuffer(len);
|
||||
if (len != 0 && !emitTxBuffer(len))
|
||||
break;
|
||||
} while (len);
|
||||
}
|
||||
}
|
||||
@@ -169,21 +170,36 @@ int32_t StreamAPI::readStream()
|
||||
/**
|
||||
* Send the current txBuffer over our stream
|
||||
*/
|
||||
void StreamAPI::emitTxBuffer(size_t len)
|
||||
bool StreamAPI::writeFrame(uint8_t *buf, size_t len)
|
||||
{
|
||||
if (len != 0) {
|
||||
txBuf[0] = START1;
|
||||
txBuf[1] = START2;
|
||||
txBuf[2] = (len >> 8) & 0xff;
|
||||
txBuf[3] = len & 0xff;
|
||||
if (len == 0 || !canWrite)
|
||||
return false;
|
||||
|
||||
auto totalLen = len + HEADER_LEN;
|
||||
// Serialize stream writes against `emitLogRecord` so a LOG_ firing
|
||||
// mid-packet-emission can't interleave bytes on the wire.
|
||||
concurrency::LockGuard guard(&streamLock);
|
||||
stream->write(txBuf, totalLen);
|
||||
buf[0] = START1;
|
||||
buf[1] = START2;
|
||||
buf[2] = (len >> 8) & 0xff;
|
||||
buf[3] = len & 0xff;
|
||||
|
||||
auto totalLen = len + HEADER_LEN;
|
||||
// Serialize write-readiness checks, writes and write-failure handling
|
||||
// against concurrent stream writes/close.
|
||||
concurrency::LockGuard guard(&streamLock);
|
||||
if (!canWriteFrame(totalLen))
|
||||
return false;
|
||||
|
||||
size_t written = stream->write(buf, totalLen);
|
||||
if (written == totalLen) {
|
||||
stream->flush();
|
||||
return true;
|
||||
}
|
||||
|
||||
onFrameWriteFailed(totalLen, written);
|
||||
return false;
|
||||
}
|
||||
|
||||
bool StreamAPI::emitTxBuffer(size_t len)
|
||||
{
|
||||
return writeFrame(txBuf, len);
|
||||
}
|
||||
|
||||
void StreamAPI::emitRebooted()
|
||||
@@ -221,20 +237,7 @@ void StreamAPI::emitLogRecord(meshtastic_LogRecord_Level level, const char *src,
|
||||
|
||||
size_t len =
|
||||
pb_encode_to_bytes(txBufLog + HEADER_LEN, meshtastic_FromRadio_size, &meshtastic_FromRadio_msg, &fromRadioScratchLog);
|
||||
if (len != 0) {
|
||||
txBufLog[0] = START1;
|
||||
txBufLog[1] = START2;
|
||||
txBufLog[2] = (len >> 8) & 0xff;
|
||||
txBufLog[3] = len & 0xff;
|
||||
|
||||
auto totalLen = len + HEADER_LEN;
|
||||
// Serialize stream writes against `emitTxBuffer` so a packet
|
||||
// emission in flight on another task doesn't interleave bytes
|
||||
// with this log record.
|
||||
concurrency::LockGuard guard(&streamLock);
|
||||
stream->write(txBufLog, totalLen);
|
||||
stream->flush();
|
||||
}
|
||||
writeFrame(txBufLog, len);
|
||||
}
|
||||
|
||||
/// Hookable to find out when connection changes
|
||||
@@ -249,4 +252,4 @@ void StreamAPI::onConnectionChanged(bool connected)
|
||||
// received a packet in a while
|
||||
powerFSM.trigger(EVENT_SERIAL_DISCONNECTED);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -80,7 +80,7 @@ class StreamAPI : public PhoneAPI
|
||||
/**
|
||||
* Send the current txBuffer over our stream
|
||||
*/
|
||||
void emitTxBuffer(size_t len);
|
||||
bool emitTxBuffer(size_t len);
|
||||
|
||||
/// Are we allowed to write packets to our output stream (subclasses can turn this off - i.e. SerialConsole)
|
||||
bool canWrite = true;
|
||||
@@ -91,7 +91,12 @@ class StreamAPI : public PhoneAPI
|
||||
/// Low level function to emit a protobuf encapsulated log record
|
||||
void emitLogRecord(meshtastic_LogRecord_Level level, const char *src, const char *format, va_list arg);
|
||||
|
||||
virtual bool canWriteFrame(size_t frameLen) { return true; }
|
||||
virtual void onFrameWriteFailed(size_t frameLen, size_t writtenLen) {}
|
||||
|
||||
private:
|
||||
bool writeFrame(uint8_t *buf, size_t len);
|
||||
|
||||
/// Dedicated scratch + tx buffer for LogRecord emission.
|
||||
///
|
||||
/// The main packet emission path (`writeStream` -> `getFromRadio` ->
|
||||
@@ -113,4 +118,4 @@ class StreamAPI : public PhoneAPI
|
||||
meshtastic_FromRadio fromRadioScratchLog = {};
|
||||
uint8_t txBufLog[MAX_STREAM_BUF_SIZE] = {0};
|
||||
concurrency::Lock streamLock;
|
||||
};
|
||||
};
|
||||
|
||||
@@ -22,12 +22,38 @@ template <typename T> void ServerAPI<T>::close()
|
||||
StreamAPI::close();
|
||||
}
|
||||
|
||||
/// Check the current underlying physical link to see if the client is currently connected
|
||||
/// Check the current underlying physical link to see if the client is currently
|
||||
/// connected
|
||||
template <typename T> bool ServerAPI<T>::checkIsConnected()
|
||||
{
|
||||
return client.connected();
|
||||
}
|
||||
|
||||
template <typename T> bool ServerAPI<T>::canWriteFrame(size_t)
|
||||
{
|
||||
// Only a dropped link is a reason to refuse a write up front. A full transmit
|
||||
// buffer (availableForWrite() == 0) is normal backpressure, not a dead socket,
|
||||
// so we must not close the connection on it. A genuinely failed write is
|
||||
// detected after the fact in onFrameWriteFailed().
|
||||
if (!client.connected()) {
|
||||
canWrite = false;
|
||||
enabled = false;
|
||||
LOG_WARN("TCP client disconnected before write, closing API service");
|
||||
close();
|
||||
return false;
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
template <typename T> void ServerAPI<T>::onFrameWriteFailed(size_t frameLen, size_t writtenLen)
|
||||
{
|
||||
canWrite = false;
|
||||
enabled = false;
|
||||
LOG_WARN("TCP client write short (%lu/%lu bytes), closing API service", (unsigned long)writtenLen, (unsigned long)frameLen);
|
||||
close();
|
||||
}
|
||||
|
||||
template <class T> int32_t ServerAPI<T>::runOnce()
|
||||
{
|
||||
if (client.connected()) {
|
||||
@@ -95,4 +121,4 @@ template <class T, class U> int32_t APIServerPort<T, U>::runOnce()
|
||||
waitTime = 100;
|
||||
#endif
|
||||
return 100; // only check occasionally for incoming connections
|
||||
}
|
||||
}
|
||||
|
||||
@@ -29,6 +29,8 @@ template <class T> class ServerAPI : public StreamAPI, private concurrency::OSTh
|
||||
/// We override this method to prevent publishing EVENT_SERIAL_CONNECTED/DISCONNECTED for wifi links (we want the board to
|
||||
/// stay in the POWERED state to prevent disabling wifi)
|
||||
virtual void onConnectionChanged(bool connected) override {}
|
||||
virtual bool canWriteFrame(size_t frameLen) override;
|
||||
virtual void onFrameWriteFailed(size_t frameLen, size_t writtenLen) override;
|
||||
|
||||
virtual int32_t runOnce() override; // Check for dropped client connections
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user