Skip to content

Commit 3c18c8d

Browse files
emelialei88emelialei88
committed
cancel reauthn timer when channel goes down
Signed-off-by: Emelia Lei <[email protected]>
1 parent 4b8c575 commit 3c18c8d

File tree

2 files changed

+72
-50
lines changed

2 files changed

+72
-50
lines changed

‎src/groups/bmq/bmqimp/bmqimp_authenticatedchannelfactory.cpp‎

Lines changed: 47 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -115,27 +115,6 @@ AuthenticatedChannelFactoryConfig::AuthenticatedChannelFactoryConfig(
115115
// ---------------------------------
116116

117117
// PRIVATE ACCESSORS
118-
void AuthenticatedChannelFactory::baseResultCallback(
119-
const ResultCallback& cb,
120-
bmqio::ChannelFactoryEvent::Enum event,
121-
const bmqio::Status& status,
122-
const bsl::shared_ptr<bmqio::Channel>& channel) const
123-
{
124-
if (event != bmqio::ChannelFactoryEvent::e_CHANNEL_UP) {
125-
cb(event, status, channel);
126-
return; // RETURN
127-
}
128-
129-
// We will skip authentication if no authentication credential
130-
// callback provided.
131-
if (d_config.d_authnCredentialCb) {
132-
authenticate(channel, cb);
133-
}
134-
else {
135-
cb(event, status, channel);
136-
}
137-
}
138-
139118
void AuthenticatedChannelFactory::sendRequest(
140119
const bsl::shared_ptr<bmqio::Channel>& channel,
141120
const ResultCallback& cb) const
@@ -227,12 +206,49 @@ void AuthenticatedChannelFactory::authenticate(
227206
readResponse(channel, cb);
228207
}
229208

209+
int AuthenticatedChannelFactory::timeoutInterval(int lifetimeMs) const
210+
{
211+
BSLS_ASSERT_SAFE(lifetimeMs >= 0);
212+
const int intervalMsWithRatio = lifetimeMs * k_REAUTHN_EARLY_RATIO;
213+
const int intervalMsWithBuffer = bsl::max(0,
214+
lifetimeMs -
215+
k_REAUTHN_EARLY_BUFFER);
216+
return bsl::min(intervalMsWithRatio, intervalMsWithBuffer);
217+
}
218+
219+
// PRIVATE MANIPULATORS
220+
void AuthenticatedChannelFactory::baseResultCallback(
221+
const ResultCallback& cb,
222+
bmqio::ChannelFactoryEvent::Enum event,
223+
const bmqio::Status& status,
224+
const bsl::shared_ptr<bmqio::Channel>& channel)
225+
{
226+
if (event != bmqio::ChannelFactoryEvent::e_CHANNEL_UP) {
227+
cb(event, status, channel);
228+
return; // RETURN
229+
}
230+
231+
channel->onClose(
232+
bdlf::BindUtil::bind(&AuthenticatedChannelFactory::onChannelDown,
233+
this,
234+
bdlf::PlaceHolders::_1)); // status
235+
236+
// We will skip authentication if no authentication credential
237+
// callback provided.
238+
if (d_config.d_authnCredentialCb) {
239+
authenticate(channel, cb);
240+
}
241+
else {
242+
cb(event, status, channel);
243+
}
244+
}
245+
230246
void AuthenticatedChannelFactory::readPacketsCb(
231247
const bsl::shared_ptr<bmqio::Channel>& channel,
232248
const ResultCallback& cb,
233249
const bmqio::Status& status,
234250
int* numNeeded,
235-
bdlbb::Blob* blob) const
251+
bdlbb::Blob* blob)
236252
{
237253
if (!status) {
238254
// Read failure.
@@ -272,7 +288,7 @@ void AuthenticatedChannelFactory::readPacketsCb(
272288
void AuthenticatedChannelFactory::onBrokerAuthenticationResponse(
273289
const bdlbb::Blob& packet,
274290
const ResultCallback& cb,
275-
const bsl::shared_ptr<bmqio::Channel>& channel) const
291+
const bsl::shared_ptr<bmqio::Channel>& channel)
276292
{
277293
BALL_LOG_TRACE << "Received a packet:\n"
278294
<< bmqu::BlobStartHexDumper(&packet);
@@ -303,20 +319,19 @@ void AuthenticatedChannelFactory::onBrokerAuthenticationResponse(
303319
cb(bmqio::ChannelFactoryEvent::e_CHANNEL_UP, bmqio::Status(), channel);
304320
}
305321

306-
int AuthenticatedChannelFactory::timeoutInterval(int lifetimeMs) const
322+
void AuthenticatedChannelFactory::onChannelDown(
323+
BSLA_UNUSED const bmqio::Status& status)
307324
{
308-
BSLS_ASSERT_SAFE(lifetimeMs >= 0);
309-
const int intervalMsWithRatio = lifetimeMs * k_REAUTHN_EARLY_RATIO;
310-
const int intervalMsWithBuffer = bsl::max(0,
311-
lifetimeMs -
312-
k_REAUTHN_EARLY_BUFFER);
313-
return bsl::min(intervalMsWithRatio, intervalMsWithBuffer);
325+
// executed by the *IO* thread
326+
327+
// Cancel pending reauthentication event when a channel goes down.
328+
d_config.d_scheduler_p->cancelEvent(&d_reauthenticationTimeoutHandle);
314329
}
315330

316331
void AuthenticatedChannelFactory::processAuthenticationEvent(
317332
const bmqp::Event& event,
318333
const ResultCallback& cb,
319-
const bsl::shared_ptr<bmqio::Channel>& channel) const
334+
const bsl::shared_ptr<bmqio::Channel>& channel)
320335
{
321336
bmqp_ctrlmsg::AuthenticationMessage response;
322337
const int rc = event.loadAuthenticationEvent(&response);
@@ -372,6 +387,7 @@ void AuthenticatedChannelFactory::processAuthenticationEvent(
372387

373388
// Pening events will be cancelled when Application stops.
374389
d_config.d_scheduler_p->scheduleEvent(
390+
&d_reauthenticationTimeoutHandle,
375391
bsls::TimeInterval(bmqsys::Time::nowMonotonicClock())
376392
.addMilliseconds(intervalMs),
377393
bdlf::BindUtil::bind(&AuthenticatedChannelFactory::sendRequest,

‎src/groups/bmq/bmqimp/bmqimp_authenticatedchannelfactory.h‎

Lines changed: 25 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -117,6 +117,10 @@ class AuthenticatedChannelFactory : public bmqio::ChannelFactory {
117117
// PRIVATE DATA
118118
Config d_config;
119119

120+
/// Handle for the scheduled reauthentication timer. An authentication
121+
/// request is sent when this event fires.
122+
bdlmt::EventScheduler::EventHandle d_reauthenticationTimeoutHandle;
123+
120124
/// Used to make sure no callback is invoked on a destroyed object.
121125
mutable bmqu::SharedResource<AuthenticatedChannelFactory> d_self;
122126

@@ -133,14 +137,6 @@ class AuthenticatedChannelFactory : public bmqio::ChannelFactory {
133137

134138
private:
135139
// PRIVATE ACCESSORS
136-
137-
/// Handle an event from our base ChannelFactory.
138-
void
139-
baseResultCallback(const ResultCallback& cb,
140-
bmqio::ChannelFactoryEvent::Enum event,
141-
const bmqio::Status& status,
142-
const bsl::shared_ptr<bmqio::Channel>& channel) const;
143-
144140
void sendRequest(const bsl::shared_ptr<bmqio::Channel>& channel,
145141
const ResultCallback& cb) const;
146142

@@ -150,22 +146,32 @@ class AuthenticatedChannelFactory : public bmqio::ChannelFactory {
150146
void authenticate(const bsl::shared_ptr<bmqio::Channel>& channel,
151147
const ResultCallback& cb) const;
152148

149+
/// Given the specified `lifetimeMs`, return the interval in milliseconds
150+
/// after which reauthentication should be performed. This interval is
151+
/// calculated with a buffer to avoid cutting too close to the actual
152+
/// expiration time.
153+
int timeoutInterval(int lifetimeMs) const;
154+
155+
// PRIVATE MANIPULATORS
156+
157+
/// Handle an event from our base ChannelFactory.
158+
void baseResultCallback(const ResultCallback& cb,
159+
bmqio::ChannelFactoryEvent::Enum event,
160+
const bmqio::Status& status,
161+
const bsl::shared_ptr<bmqio::Channel>& channel);
162+
153163
void readPacketsCb(const bsl::shared_ptr<bmqio::Channel>& channel,
154164
const ResultCallback& cb,
155165
const bmqio::Status& status,
156166
int* numNeeded,
157-
bdlbb::Blob* blob) const;
167+
bdlbb::Blob* blob);
158168

159169
void onBrokerAuthenticationResponse(
160170
const bdlbb::Blob& packet,
161171
const ResultCallback& cb,
162-
const bsl::shared_ptr<bmqio::Channel>& channel) const;
172+
const bsl::shared_ptr<bmqio::Channel>& channel);
163173

164-
/// Given the specified `lifetimeMs`, return the interval in milliseconds
165-
/// after which reauthentication should be performed. This interval is
166-
/// calculated with a buffer to avoid cutting too close to the actual
167-
/// expiration time.
168-
int timeoutInterval(int lifetimeMs) const;
174+
void onChannelDown(const bmqio::Status& status);
169175

170176
public:
171177
// CREATORS
@@ -186,10 +192,10 @@ class AuthenticatedChannelFactory : public bmqio::ChannelFactory {
186192
const bmqio::ConnectOptions& options,
187193
const ResultCallback& cb) BSLS_KEYWORD_OVERRIDE;
188194

189-
void processAuthenticationEvent(
190-
const bmqp::Event& event,
191-
const ResultCallback& cb,
192-
const bsl::shared_ptr<bmqio::Channel>& channel) const;
195+
void
196+
processAuthenticationEvent(const bmqp::Event& event,
197+
const ResultCallback& cb,
198+
const bsl::shared_ptr<bmqio::Channel>& channel);
193199
};
194200

195201
} // close package namespace

0 commit comments

Comments
 (0)