espressif/openthread
1
0
Fork 0
mirror of https://github.com/espressif/openthread.git synced 2026-09-12 12:10:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

[coap] enhance PendingRequest and move request management functionality (#12527)

Browse Source 
This commit enhances the `PendingRequest` class by moving more functionality
regarding the management of pending CoAP requests into this nested class
within `CoapBase`.

Specifically, this change moves the handling of the retransmission timer,
retransmission of requests, finalizing a request, and aborting all or a
subset of requests. This helps organize the code and simplifies the
`CoapBase` class implementation.
This commit is contained in:
Abtin Keshavarzian
2026-02-23 10:27:29 -06:00
committed by GitHub
parent d7000862e8
commit 5d92992b78
2 changed files with 150 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/coap/coap.cpp
+96 -52
View File
@@ -120,11 +120,11 @@ void Msg::UpdateMessageId(uint16_t aMessageId)
CoapBase::CoapBase(Instance &aInstance, Sender aSender)
: InstanceLocator(aInstance)
, mMessageId(Random::NonCrypto::GetUint16())
, mRetransmissionTimer(aInstance, Coap::HandleRetransmissionTimer, this)
, mPendingRequests(aInstance, *this)
, mResponseCache(aInstance)
, mResourceHandler(nullptr)
, mSender(aSender)
, mMessageId(Random::NonCrypto::GetUint16())
#if OPENTHREAD_CONFIG_COAP_BLOCKWISE_TRANSFER_ENABLE
, mLastResponse(nullptr)
#endif
@@ -133,22 +133,8 @@ CoapBase::CoapBase(Instance &aInstance, Sender aSender)
void CoapBase::ClearAllRequestsAndResponses(void)
{
ClearRequests(nullptr); // Clear requests matching any address.
mPendingRequests.AbortAllRequests();
mResponseCache.RemoveAll();
mRetransmissionTimer.Stop();
}
void CoapBase::ClearRequests(const Ip6::Address &aAddress) { ClearRequests(&aAddress); }
void CoapBase::ClearRequests(const Ip6::Address *aAddress)
{
for (Request &request : mPendingRequests)
{
if ((aAddress == nullptr) || (request.mMetadata.mSourceAddress == *aAddress))
{
FinalizeRequest(request, kErrorAbort);
}
}
}
void CoapBase::AddResource(Resource &aResource) { IgnoreError(mResources.Add(aResource)); }
@@ -308,7 +294,6 @@ Error CoapBase::SendMessage(Message &aMessage,
#endif
SuccessOrExit(error = mPendingRequests.AddClone(txMsg.mMessage, copyLength, request));
mRetransmissionTimer.FireAtIfEarlier(request.mMetadata.mTimerFireTime);
}
SuccessOrExit(error = Send(txMsg.mMessage, txMsg.mMessageInfo));
@@ -479,17 +464,21 @@ exit:
return error;
}
void CoapBase::HandleRetransmissionTimer(Timer &aTimer)
void CoapBase::PendingRequests::HandleTimer(Timer &aTimer)
{
static_cast<Coap *>(static_cast<TimerMilliContext &>(aTimer).GetContext())->HandleRetransmissionTimer();
static_cast<PendingRequests *>(static_cast<TimerMilliContext &>(aTimer).GetContext())->HandleTimer();
}
void CoapBase::HandleRetransmissionTimer(void)
void CoapBase::PendingRequests::HandleTimer(void)
{
NextFireTime nextTime;
for (Request &request : mPendingRequests)
for (Message &message : mRequestMessages)
{
Request request;
request.InitFrom(message);
#if OPENTHREAD_CONFIG_COAP_OBSERVE_API_ENABLE
if (request.mMetadata.IsObserveSubscription())
{
@@ -520,19 +509,19 @@ void CoapBase::HandleRetransmissionTimer(void)
nextTime.UpdateIfEarlier(request.mMetadata.mTimerFireTime);
}
mRetransmissionTimer.FireAt(nextTime);
mTimer.FireAt(nextTime);
}
void CoapBase::FinalizeRequest(Request &aRequest, Error aResult)
void CoapBase::PendingRequests::FinalizeRequest(Request &aRequest, Error aResult)
{
FinalizeRequest(aRequest, aResult, /* aResponse */ nullptr);
}
void CoapBase::FinalizeRequest(Request &aRequest, Error aResult, Msg *aResponse)
void CoapBase::PendingRequests::FinalizeRequest(Request &aRequest, Error aResult, Msg *aResponse)
{
VerifyOrExit(aRequest.HasMessage());
mPendingRequests.Remove(aRequest);
Remove(aRequest);
aRequest.mMetadata.mCallbacks.InvokeResponseHandler(aResponse, aResult);
exit:
@@ -541,18 +530,7 @@ exit:
Error CoapBase::AbortTransaction(ResponseHandler aHandler, void *aContext)
{
Error error = kErrorNotFound;
for (Request &request : mPendingRequests)
{
if (request.mMetadata.mCallbacks.Matches(aHandler, aContext))
{
FinalizeRequest(request, kErrorAbort);
error = kErrorNone;
}
}
return error;
return mPendingRequests.AbortRequestsMatching(aHandler, aContext);
}
void CoapBase::GetRequestAndCachedResponsesQueueInfo(MessageQueue::Info &aQueueInfo) const
@@ -575,6 +553,8 @@ Error CoapBase::PendingRequests::AddClone(const Message &aMessage, uint16_t aCop
mRequestMessages.Enqueue(*aRequest.mMessage);
mTimer.FireAtIfEarlier(aRequest.mMetadata.mTimerFireTime);
exit:
FreeAndNullMessageOnError(aRequest.mMessage, error);
return error;
@@ -590,7 +570,7 @@ exit:
return;
}
void CoapBase::RetransmitRequest(const Request &aRequest)
void CoapBase::PendingRequests::RetransmitRequest(const Request &aRequest)
{
Error error;
Message *clone;
@@ -601,7 +581,7 @@ void CoapBase::RetransmitRequest(const Request &aRequest)
aRequest.mMetadata.CopyInfoTo(messageInfo);
SuccessOrExit(error = Send(*clone, messageInfo));
SuccessOrExit(error = mCoapBase.Send(*clone, messageInfo));
exit:
FreeMessageOnError(clone, error);
@@ -720,7 +700,7 @@ void CoapBase::ProcessReceivedResponse(Msg &aRxMsg)
case kTypeReset:
if (aRxMsg.IsEmpty())
{
FinalizeRequest(request, kErrorAbort);
mPendingRequests.FinalizeRequest(request, kErrorAbort);
}
// Silently ignore non-empty reset messages (RFC 7252, Section 4.2).
@@ -739,7 +719,7 @@ void CoapBase::ProcessReceivedResponse(Msg &aRxMsg)
// as if it were a piggy-backed response so we can stop
// re-sending and the application can move on.
FinalizeRequest(request, kErrorNone, &aRxMsg);
mPendingRequests.FinalizeRequest(request, kErrorNone, &aRxMsg);
}
else
#endif
@@ -780,7 +760,7 @@ void CoapBase::ProcessReceivedResponse(Msg &aRxMsg)
#if OPENTHREAD_CONFIG_COAP_BLOCKWISE_TRANSFER_ENABLE
SuccessOrExit(error = ProcessBlockwiseResponse(aRxMsg, request));
#else
FinalizeRequest(request, kErrorNone, &aRxMsg);
mPendingRequests.FinalizeRequest(request, kErrorNone, &aRxMsg);
#endif
}
}
@@ -828,7 +808,7 @@ void CoapBase::ProcessReceivedResponse(Msg &aRxMsg)
}
else
{
FinalizeRequest(request, kErrorNone, &aRxMsg);
mPendingRequests.FinalizeRequest(request, kErrorNone, &aRxMsg);
}
break;
@@ -1023,7 +1003,7 @@ Error CoapBase::ProcessBlockwiseResponse(Msg &aRxMsg, Request &aRequest)
{
case 0:
// Piggybacked response.
FinalizeRequest(aRequest, kErrorNone, &aRxMsg);
mPendingRequests.FinalizeRequest(aRequest, kErrorNone, &aRxMsg);
break;
case 1: // Block1 option
if (aRxMsg.GetCode() == kCodeContinue && aRequest.mMetadata.mCallbacks.HasBlockwiseTransmitHook())
@@ -1034,7 +1014,7 @@ Error CoapBase::ProcessBlockwiseResponse(Msg &aRxMsg, Request &aRequest)
if (aRxMsg.GetCode() != kCodeContinue || !aRequest.mMetadata.mCallbacks.HasBlockwiseTransmitHook() ||
error != kErrorNone)
{
FinalizeRequest(aRequest, error, &aRxMsg);
mPendingRequests.FinalizeRequest(aRequest, error, &aRxMsg);
}
break;
case 2: // Block2 option
@@ -1046,7 +1026,7 @@ Error CoapBase::ProcessBlockwiseResponse(Msg &aRxMsg, Request &aRequest)
if (aRxMsg.GetCode() >= kCodeBadRequest || !aRequest.mMetadata.mCallbacks.HasBlockwiseReceiveHook() ||
error != kErrorNone)
{
FinalizeRequest(aRequest, error, &aRxMsg);
mPendingRequests.FinalizeRequest(aRequest, error, &aRxMsg);
}
break;
case 3: // Block1 & Block2 option
@@ -1055,11 +1035,11 @@ Error CoapBase::ProcessBlockwiseResponse(Msg &aRxMsg, Request &aRequest)
error = SendNextBlock2Request(aRequest, aRxMsg, totalTransferSize, true);
}
FinalizeRequest(aRequest, error, &aRxMsg);
mPendingRequests.FinalizeRequest(aRequest, error, &aRxMsg);
break;
default:
error = kErrorAbort;
FinalizeRequest(aRequest, error, &aRxMsg);
mPendingRequests.FinalizeRequest(aRequest, error, &aRxMsg);
break;
}
@@ -1277,7 +1257,7 @@ Error CoapBase::SendNextBlock1Request(Request &aRequest, Msg &aRxMsg)
// Conclude block-wise transfer if last block has been received
if (!requestBlockInfo.mMoreBlocks)
{
FinalizeRequest(aRequest, kErrorNone, &aRxMsg);
mPendingRequests.FinalizeRequest(aRequest, kErrorNone, &aRxMsg);
ExitNow();
}
@@ -1343,7 +1323,7 @@ Error CoapBase::SendNextBlock2Request(Request &aRequest, Msg &aRxMsg, uint32_t a
if (!msgBlockInfo.mMoreBlocks)
{
FinalizeRequest(aRequest, kErrorNone, &aRxMsg);
mPendingRequests.FinalizeRequest(aRequest, kErrorNone, &aRxMsg);
ExitNow();
}
@@ -1569,7 +1549,7 @@ Error CoapBase::ProcessObserveSend(Msg &aTxMsg, Request &aRequest)
if (mPendingRequests.FindRelatedRequest(aTxMsg, request) == kErrorNone)
{
FinalizeRequest(request, kErrorNone);
mPendingRequests.FinalizeRequest(request, kErrorNone);
}
}
}
@@ -1686,6 +1666,70 @@ bool CoapBase::Request::Metadata::IsObserveSubscription(void) const
}
#endif
CoapBase::PendingRequests::PendingRequests(Instance &aInstance, CoapBase &aCoapBase)
: mCoapBase(aCoapBase)
, mTimer(aInstance, HandleTimer, this)
{
}
void CoapBase::PendingRequests::AbortAllRequests(void)
{
IgnoreError(AbortAllMatching(Matcher()));
mTimer.Stop();
}
void CoapBase::PendingRequests::AbortRequestsMatching(const Ip6::Address &aAddress)
{
IgnoreError(AbortAllMatching(Matcher(aAddress)));
}
Error CoapBase::PendingRequests::AbortRequestsMatching(ResponseHandler aHandler, void *aContext)
{
return AbortAllMatching(Matcher(aHandler, aContext));
}
Error CoapBase::PendingRequests::AbortAllMatching(const Matcher &aMatcher)
{
Error error = kErrorNotFound;
for (Message &message : mRequestMessages)
{
Request request;
request.InitFrom(message);
if (aMatcher.Matches(request))
{
FinalizeRequest(request, kErrorAbort);
error = kErrorNone;
}
}
return error;
}
bool CoapBase::PendingRequests::Matcher::Matches(const Request &aRequest) const
{
bool matches = false;
switch (mMode)
{
case kAny:
break;
case kAddress:
VerifyOrExit(aRequest.mMetadata.mSourceAddress == *mAddress);
break;
case kHandler:
VerifyOrExit(aRequest.mMetadata.mCallbacks.Matches(mHandler, mContext));
break;
}
matches = true;
exit:
return matches;
}
//---------------------------------------------------------------------------------------------------------------------
// CoapBase::ResponseCache
src/core/coap/coap.hpp
+54 -60
View File
@@ -312,7 +312,7 @@ public:
*
* @param[in] aAddress A reference to the specified address.
*/
void ClearRequests(const Ip6::Address &aAddress);
void ClearRequests(const Ip6::Address &aAddress) { mPendingRequests.AbortRequestsMatching(aAddress); }
/**
* Adds a resource to the CoAP server.
@@ -834,45 +834,52 @@ private:
struct Iterator;
public:
explicit PendingRequests(Instance &aInstance, CoapBase &aCoapBase);
Error AddClone(const Message &aMessage, uint16_t aCopyLength, Request &aRequest);
void Remove(Request &aRequest);
Error FindRelatedRequest(const Msg &aMsg, Request &aRequest);
void FinalizeRequest(Request &aRequest, Error aResult);
void FinalizeRequest(Request &aRequest, Error aResult, Msg *aResponse);
void AbortAllRequests(void);
void AbortRequestsMatching(const Ip6::Address &aAddress);
Error AbortRequestsMatching(ResponseHandler aHandler, void *aContext);
void GetInfo(MessageQueue::Info &aInfo) const { mRequestMessages.GetInfo(aInfo); }
// Similar to the MessageQueue iterator, as we iterate over
// the PendingRequests entries, the current request can
// be safely removed without invalidating the iterator.
Iterator begin(void) { return Iterator(mRequestMessages.GetHead()); }
Iterator end(void) { return Iterator(); }
private:
struct Iterator : public Unequatable<Iterator>
class Matcher
{
Iterator(void)
: mMessageIterator()
public:
using Handler = ResponseHandler;
Matcher(void) { mMode = kAny; }
Matcher(const Ip6::Address &aSourceAddress) { mMode = kAddress, mAddress = &aSourceAddress; }
Matcher(Handler aHandler, void *aContext) { mMode = kHandler, mHandler = aHandler, mContext = aContext; }
bool Matches(const Request &aRequest) const;
private:
enum MatchMode
{
}
kAny,
kAddress,
kHandler,
};
explicit Iterator(Message *aMessage)
: mMessageIterator(aMessage)
{
}
void operator++(void) { mMessageIterator++; }
void operator++(int) { mMessageIterator++; }
bool operator==(const Iterator &aOther) const { return mMessageIterator == aOther.mMessageIterator; }
Request &operator*(void)
{
mRequest.InitFrom(*mMessageIterator);
return mRequest;
}
Message::Iterator mMessageIterator;
Request mRequest;
MatchMode mMode;
const Ip6::Address *mAddress;
Handler mHandler;
void *mContext;
};
MessageQueue mRequestMessages;
Error AbortAllMatching(const Matcher &aMatcher);
void RetransmitRequest(const Request &aRequest);
static void HandleTimer(Timer &aTimer);
void HandleTimer(void);
CoapBase &mCoapBase;
MessageQueue mRequestMessages;
TimerMilliContext mTimer;
};
class ResponseCache
@@ -905,23 +912,17 @@ private:
TimerMilliContext mTimer;
};
Message *InitMessage(Message *aMessage, Type aType, Uri aUri);
Message *InitResponse(Message *aMessage, const Message &aRequest);
static void HandleRetransmissionTimer(Timer &aTimer);
void HandleRetransmissionTimer(void);
void ClearRequests(const Ip6::Address *aAddress);
void FinalizeRequest(Request &aRequest, Error aResult);
void FinalizeRequest(Request &aRequest, Error aResult, Msg *aResponse);
bool InvokeResponseFallback(Msg &aRxMsg) const;
void ProcessReceivedRequest(Msg &aRxMsg);
void ProcessReceivedResponse(Msg &aRxMsg);
Error SendMessage(Message &aMessage,
const Ip6::MessageInfo &aMessageInfo,
const TxParameters *aTxParameters,
const SendCallbacks &aCallbacks);
void RetransmitRequest(const Request &aRequest);
Error SendEmptyMessage(Type aType, const Msg &aRxMsg);
Error Send(ot::Message &aMessage, const Ip6::MessageInfo &aMessageInfo);
Message *InitMessage(Message *aMessage, Type aType, Uri aUri);
Message *InitResponse(Message *aMessage, const Message &aRequest);
bool InvokeResponseFallback(Msg &aRxMsg) const;
void ProcessReceivedRequest(Msg &aRxMsg);
void ProcessReceivedResponse(Msg &aRxMsg);
Error SendMessage(Message &aMessage,
const Ip6::MessageInfo &aMessageInfo,
const TxParameters *aTxParameters,
const SendCallbacks &aCallbacks);
Error SendEmptyMessage(Type aType, const Msg &aRxMsg);
Error Send(ot::Message &aMessage, const Ip6::MessageInfo &aMessageInfo);
#if OPENTHREAD_CONFIG_COAP_BLOCKWISE_TRANSFER_ENABLE
@@ -947,22 +948,15 @@ private:
Error ProcessObserveSend(Msg &aTxMsg, Request &aRequest);
#endif
PendingRequests mPendingRequests;
uint16_t mMessageId;
TimerMilliContext mRetransmissionTimer;
LinkedList<Resource> mResources;
Callback<Interceptor> mInterceptor;
ResponseCache mResponseCache;
PendingRequests mPendingRequests;
ResponseCache mResponseCache;
LinkedList<Resource> mResources;
Callback<Interceptor> mInterceptor;
Callback<RequestHandler> mDefaultHandler;
Callback<ResponseFallback> mResponseFallback;
ResourceHandler mResourceHandler;
const Sender mSender;
ResourceHandler mResourceHandler;
Sender mSender;
uint16_t mMessageId;
#if OPENTHREAD_CONFIG_COAP_BLOCKWISE_TRANSFER_ENABLE
LinkedList<ResourceBlockWise> mBlockWiseResources;
Message *mLastResponse;