mqbnet_initialconnectioncontext.cpp

// Copyright 2025 Bloomberg Finance L.P.
// SPDX-License-Identifier: Apache-2.0
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include <mqbnet_initialconnectioncontext.h>

#include <mqbscm_version.h>

// MQB
#include <mqbcfg_messages.h>
#include <mqbnet_authenticationcontext.h>
#include <mqbnet_negotiationcontext.h>

// BMQ
#include <bmqio_channel.h>
#include <bmqio_channelutil.h>
#include <bmqp_ctrlmsg_messages.h>
#include <bmqp_event.h>
#include <bmqu_blob.h>
#include <bmqu_memoutstream.h>

// BDE
#include <ball_log.h>
#include <bdlb_print.h>
#include <bdlb_string.h>
#include <bdlma_localsequentialallocator.h>
#include <bsl_memory.h>
#include <bsl_ostream.h>
#include <bsl_vector.h>

namespace BloombergLP {
namespace mqbnet {

namespace {

const int k_INITIALCONNECTION_READTIMEOUT = 3 * 60;  // 3 minutes

}

// -----------------------------
// struct InitialConnectionState
// -----------------------------

bsl::ostream& InitialConnectionState::print(bsl::ostream& stream,
                                            InitialConnectionState::Enum value,
                                            int                          level,
                                            int spacesPerLevel)
{
    if (stream.bad()) {
        return stream;  // RETURN
    }

    bdlb::Print::indent(stream, level, spacesPerLevel);
    stream << InitialConnectionState::toAscii(value);

    if (spacesPerLevel >= 0) {
        stream << '\n';
    }

    return stream;
}

const char* InitialConnectionState::toAscii(InitialConnectionState::Enum value)
{
#define CASE(X)                                                               \
    case e_##X: return #X;

    switch (value) {
        CASE(INITIAL)
        CASE(AUTHENTICATING)
        CASE(AUTHENTICATED)
        CASE(ANON_AUTHENTICATING)
        CASE(NEGOTIATING_OUTBOUND)
        CASE(NEGOTIATED)
        CASE(FAILED)
    default: return "(* UNKNOWN *)";
    }

#undef CASE
}

bool InitialConnectionState::fromAscii(InitialConnectionState::Enum* out,
                                       const bslstl::StringRef&      str)
{
#define CHECKVALUE(M)                                                         \
    if (bdlb::String::areEqualCaseless(                                       \
            toAscii(InitialConnectionState::e_##M),                           \
            str.data(),                                                       \
            static_cast<int>(str.length()))) {                                \
        *out = InitialConnectionState::e_##M;                                 \
        return true;                                                          \
    }

    CHECKVALUE(INITIAL)
    CHECKVALUE(AUTHENTICATING)
    CHECKVALUE(AUTHENTICATED)
    CHECKVALUE(ANON_AUTHENTICATING)
    CHECKVALUE(NEGOTIATING_OUTBOUND)
    CHECKVALUE(NEGOTIATED)
    CHECKVALUE(FAILED)

    // Invalid string
    return false;

#undef CHECKVALUE
}

// -----------------------------
// struct InitialConnectionEvent
// -----------------------------

bsl::ostream& InitialConnectionEvent::print(bsl::ostream& stream,
                                            InitialConnectionEvent::Enum value,
                                            int                          level,
                                            int spacesPerLevel)
{
    if (stream.bad()) {
        return stream;  // RETURN
    }

    bdlb::Print::indent(stream, level, spacesPerLevel);
    stream << InitialConnectionEvent::toAscii(value);

    if (spacesPerLevel >= 0) {
        stream << '\n';
    }

    return stream;
}

const char* InitialConnectionEvent::toAscii(InitialConnectionEvent::Enum value)
{
#define CASE(X)                                                               \
    case e_##X: return #X;

    switch (value) {
        CASE(NONE)
        CASE(OUTBOUND_NEGOTIATION)
        CASE(INCOMING)
        CASE(AUTHN_REQUEST)
        CASE(NEGOTIATION_MESSAGE)
        CASE(AUTHN_SUCCESS)
        CASE(ERROR)
    default: return "(* UNKNOWN *)";
    }

#undef CASE
}

bool InitialConnectionEvent::fromAscii(InitialConnectionEvent::Enum* out,
                                       const bslstl::StringRef&      str)
{
#define CHECKVALUE(M)                                                         \
    if (bdlb::String::areEqualCaseless(                                       \
            toAscii(InitialConnectionEvent::e_##M),                           \
            str.data(),                                                       \
            static_cast<int>(str.length()))) {                                \
        *out = InitialConnectionEvent::e_##M;                                 \
        return true;                                                          \
    }

    CHECKVALUE(NONE)
    CHECKVALUE(OUTBOUND_NEGOTIATION)
    CHECKVALUE(INCOMING)
    CHECKVALUE(AUTHN_REQUEST)
    CHECKVALUE(NEGOTIATION_MESSAGE)
    CHECKVALUE(AUTHN_SUCCESS)
    CHECKVALUE(ERROR)

    // Invalid string
    return false;

#undef CHECKVALUE
}

// -------------------------
// class NegotiationUserData
// -------------------------

NegotiationUserData::~NegotiationUserData()
{
    // NOTHING: Pure interface
}

// ------------------------------
// class InitialConnectionContext
// ------------------------------

// CREATORS
InitialConnectionContext::InitialConnectionContext(
    bool                                        isIncoming,
    mqbnet::Authenticator*                      authenticator,
    mqbnet::Negotiator*                         negotiator,
    const bsl::shared_ptr<NegotiationUserData>& userData,
    void*                                       resultState,
    const bsl::shared_ptr<bmqio::Channel>&      channel,
    const InitialConnectionCompleteCb&          initialConnectionCompleteCb,
    bslma::Allocator*                           allocator)
: d_allocator_p(allocator)
, d_mutex()
, d_authenticator_p(authenticator)
, d_negotiator_p(negotiator)
, d_resultState_p(resultState)
, d_userData_sp(userData)
, d_channel_sp(channel)
, d_authenticationCtx_sp()
, d_negotiationCtx_sp()
, d_initialConnectionCompleteCb(initialConnectionCompleteCb)
, d_authenticationEncodingType(bmqp::EncodingType::e_BER)
, d_state(InitialConnectionState::e_INITIAL)
, d_isIncoming(isIncoming)
, d_isClosed(false)

{
    // NOTHING
}

InitialConnectionContext::~InitialConnectionContext()
{
    // NOTHING
}

// PRIVATE MANIPULATORS
void InitialConnectionContext::setState(InitialConnectionState::Enum value,
                                        InitialConnectionEvent::Enum event)
{
    BALL_LOG_DEBUG << "State transition: " << d_state << " -> (" << event
                   << ") -> " << value << " [peer: " << d_channel_sp.get()
                   << "]";
    d_state = value;
}

int InitialConnectionContext::scheduleRead(bsl::ostream& errorDescription)
{
    // Schedule a TimedRead
    bmqio::Status status;
    channel()->read(
        &status,
        bmqp::Protocol::k_PACKET_MIN_SIZE,
        bdlf::BindUtil::bind(&InitialConnectionContext::readCallback,
                             this,
                             bdlf::PlaceHolders::_1,   // status
                             bdlf::PlaceHolders::_2,   // numNeeded
                             bdlf::PlaceHolders::_3),  // blob
        bsls::TimeInterval(k_INITIALCONNECTION_READTIMEOUT));

    if (!status) {
        errorDescription << "Read failed while negotiating: " << status;
        return -1;  // RETURN
    }

    return 0;
}

int InitialConnectionContext::readBlob(bsl::ostream& errorDescription,
                                       bdlbb::Blob*  outPacket,
                                       bool*         isFullBlob,
                                       int*          numNeeded,
                                       bdlbb::Blob*  blob)
{
    int rc = bmqio::ChannelUtil::handleRead(outPacket, numNeeded, blob);
    if (rc != 0) {
        // This indicates a non recoverable error...
        errorDescription << "Unrecoverable read error:\n"
                         << bmqu::BlobStartHexDumper(blob);
        return -1;  // RETURN
    }

    if (outPacket->length() == 0) {
        // Don't yet have a full blob
        *isFullBlob = false;
        return 0;  // RETURN
    }

    // Have a full blob, indicate no more bytes needed (we have to do this
    // because 'handleRead' above set it back to 4 at the end).
    *isFullBlob = true;
    *numNeeded  = 0;

    return 0;
}

int InitialConnectionContext::processBlob(bsl::ostream&      errorDescription,
                                          const bdlbb::Blob& blob)
{
    // executed by one of the *IO* threads

    bsl::variant<bsl::monostate,
                 bmqp_ctrlmsg::AuthenticationMessage,
                 bmqp_ctrlmsg::NegotiationMessage>
        message;

    int rc = decodeInitialConnectionMessage(errorDescription, &message, blob);

    if (rc != 0) {
        errorDescription << "Failed to decode initial connection message";
        return -1;  // RETURN
    }

    if (bsl::holds_alternative<bsl::monostate>(message)) {
        errorDescription << "Decode AuthenticationMessage or "
                            "NegotiationMessage succeeds but nothing gets "
                            "loaded in.";
        return -1;
    }
    else if (bsl::holds_alternative<bmqp_ctrlmsg::AuthenticationMessage>(
                 message)) {
        handleEvent(bsl::string(),
                    InitialConnectionEvent::e_AUTHN_REQUEST,
                    message);
    }
    else {
        handleEvent(bsl::string(),
                    InitialConnectionEvent::e_NEGOTIATION_MESSAGE,
                    message);
    }

    return 0;
}

int InitialConnectionContext::decodeInitialConnectionMessage(
    bsl::ostream&                                   errorDescription,
    bsl::variant<bsl::monostate,
                 bmqp_ctrlmsg::AuthenticationMessage,
                 bmqp_ctrlmsg::NegotiationMessage>* message,
    const bdlbb::Blob&                              blob)
{
    BSLS_ASSERT(message);

    enum RcEnum {
        // Value for the various RC error categories
        rc_SUCCESS                      = 0,
        rc_INVALID_MESSAGE              = -1,
        rc_INVALID_EVENT                = -2,
        rc_INVALID_AUTHENTICATION_EVENT = -3,
        rc_INVALID_CONTROL_EVENT        = -4
    };

    bdlma::LocalSequentialAllocator<2048> localAllocator(d_allocator_p);

    bmqp::Event event(&blob, &localAllocator);

    if (!event.isValid()) {
        errorDescription << "Invalid negotiation message received "
                         << "(packet is not a valid BlazingMQ event):\n"
                         << bmqu::BlobStartHexDumper(&blob);
        return rc_INVALID_MESSAGE;  // RETURN
    }

    bmqp_ctrlmsg::AuthenticationMessage authenticationMessage;
    bmqp_ctrlmsg::NegotiationMessage    negotiationMessage;

    if (event.isAuthenticationEvent()) {
        BALL_LOG_DEBUG << "Received AuthenticationEvent: "
                       << bmqu::BlobStartHexDumper(&blob);
        const int rc = event.loadAuthenticationEvent(&authenticationMessage);
        if (rc != 0) {
            errorDescription
                << "Invalid message received [reason: 'authentication "
                   "event is not an AuthenticationMessage', rc: "
                << rc << "]:" << bmqu::BlobStartHexDumper(&blob);
            return rc_INVALID_AUTHENTICATION_EVENT;  // RETURN
        }

        d_authenticationEncodingType = event.authenticationEventEncodingType();
        *message                     = authenticationMessage;
    }
    else if (event.isControlEvent()) {
        BALL_LOG_DEBUG << "Received ControlEvent: "
                       << bmqu::BlobStartHexDumper(&blob);
        const int rc = event.loadControlEvent(&negotiationMessage);
        if (rc != 0) {
            errorDescription << "Invalid message received [reason: 'control "
                                "event is not a NegotiationMessage', rc: "
                             << rc << "]:" << bmqu::BlobStartHexDumper(&blob);
            return rc_INVALID_CONTROL_EVENT;  // RETURN
        }

        *message = negotiationMessage;
    }
    else {
        errorDescription
            << "Invalid initial connection message received "
            << "(packet is not an AuthenticationEvent or ControlEvent):\n"
            << bmqu::BlobStartHexDumper(&blob);
        return rc_INVALID_EVENT;  // RETURN
    }

    return rc_SUCCESS;
}

void InitialConnectionContext::createNegotiationContext()
{
    if (d_negotiationCtx_sp) {
        return;  // RETURN
    }

    d_negotiationCtx_sp = bsl::allocate_shared<mqbnet::NegotiationContext>(
        d_allocator_p,
        this  // initialConnectionContext
    );
}

int InitialConnectionContext::handleAnonAuthentication(
    bsl::ostream& errorDescription)
{
    // executed by one of the *IO* threads

    // PRECONDITIONS
    BSLS_ASSERT_SAFE(!authenticationContext());

    if (!d_authenticator_p->anonymousCredential()) {
        errorDescription << "Anonymous credential is disallowed, "
                         << "cannot negotiate without authentication.";
        return -1;  // RETURN
    }

    bmqp_ctrlmsg::AuthenticationMessage  authenticationMessage;
    bmqp_ctrlmsg::AuthenticationRequest& authenticationRequest =
        authenticationMessage.makeAuthenticationRequest();

    const mqbcfg::Credential& anonymousCredential =
        d_authenticator_p->anonymousCredential().value();
    authenticationRequest.mechanism() = anonymousCredential.mechanism();
    authenticationRequest.data()      = bsl::vector<char>(
        anonymousCredential.identity().begin(),
        anonymousCredential.identity().end());

    const int rc = d_authenticator_p->handleAuthentication(
        errorDescription,
        this,
        authenticationMessage);

    return rc;
}

// MANIPULATORS
void InitialConnectionContext::setResultState(void* value)
{
    d_resultState_p = value;
}

void InitialConnectionContext::setAuthenticationContext(
    const bsl::shared_ptr<AuthenticationContext>& value)
{
    // PRECONDITIONS
    BSLS_ASSERT_SAFE(!d_authenticationCtx_sp);

    d_authenticationCtx_sp = value;
}

void InitialConnectionContext::onClose()
{
    d_isClosed = true;

    // Propagate close
    if (d_authenticationCtx_sp) {
        d_authenticationCtx_sp->onClose();
    }
}

void InitialConnectionContext::readCallback(const bmqio::Status& status,
                                            int*                 numNeeded,
                                            bdlbb::Blob*         blob)
{
    // executed by one of the *IO* threads

    // PRECONDITIONS

    BALL_LOG_TRACE << "InitialConnectionContext readCb: [status: " << status
                   << ", peer: '" << channel().get() << "']";

    bmqu::MemOutStream errStream;

    if (!status) {
        errStream << "Read error: " << status;
        handleEvent(errStream.str(), InitialConnectionEvent::e_ERROR);
        return;  // RETURN
    }

    bdlbb::Blob outPacket;
    bool        isFullBlob = true;
    int rc = readBlob(errStream, &outPacket, &isFullBlob, numNeeded, blob);
    if (rc != 0) {
        handleEvent(errStream.str(), InitialConnectionEvent::e_ERROR);
        return;  // RETURN
    }

    if (!isFullBlob) {
        return;  // RETURN
    }

    rc = processBlob(errStream, outPacket);
    if (rc != 0) {
        handleEvent(errStream.str(), InitialConnectionEvent::e_ERROR);
        return;  // RETURN
    }
}

void InitialConnectionContext::handleInitialConnection()
{
    if (!isIncoming()) {
        // TODO: When we are ready to move on to the next step, we should
        // call `authenticationOutbound` here instead before calling
        // `negotiateOutbound`.
        handleEvent(bsl::string(),
                    mqbnet::InitialConnectionEvent::e_OUTBOUND_NEGOTIATION);
    }
    else {
        handleEvent(bsl::string(), mqbnet::InitialConnectionEvent::e_INCOMING);
    }
}

void InitialConnectionContext::handleEvent(
    const bsl::string&                                    errorDescription,
    InitialConnectionEvent::Enum                          event,
    const bsl::variant<bsl::monostate,
                       bmqp_ctrlmsg::AuthenticationMessage,
                       bmqp_ctrlmsg::NegotiationMessage>& message)
{
    // executed by an *AUTHENTICATION* or one of the *IO* threads

    enum RcEnum {
        // Value for the various RC error categories
        rc_SUCCESS = 0,
        rc_ERROR   = -1
    };

    bmqu::MemOutStream errStream(d_allocator_p);
    int                rc = rc_ERROR;

    bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCKED

    BALL_LOG_DEBUG << "Enter InitialConnectionContext::handleEvent: "
                   << "state = " << d_state << ", event = " << event
                   << " [peer: " << d_channel_sp.get() << "]";

    InitialConnectionState::Enum oldState = d_state;

    switch (event) {
    case InitialConnectionEvent::e_OUTBOUND_NEGOTIATION: {
        if (oldState == InitialConnectionState::e_INITIAL) {
            setState(InitialConnectionState::e_NEGOTIATING_OUTBOUND, event);

            createNegotiationContext();

            rc = d_negotiator_p->negotiateOutbound(errStream, this);
            if (rc == rc_SUCCESS) {
                rc = scheduleRead(errStream);
            }
        }
        else {
            errStream << "Unexpected event received: " << oldState << " -> "
                      << event;
            BALL_LOG_ERROR << "#UNEXPECTED_STATE " << errStream.str()
                           << " [peer: " << channel().get() << "]";
        }
        break;
    }
    case InitialConnectionEvent::e_INCOMING: {
        if (oldState == InitialConnectionState::e_INITIAL) {
            // For incoming connections, start reading the first message
            rc = scheduleRead(errStream);
        }
        else {
            errStream << "Unexpected event received: " << oldState << " -> "
                      << event;
            BALL_LOG_ERROR << "#UNEXPECTED_STATE " << errStream.str()
                           << " [peer: " << channel().get() << "]";
        }
        break;
    }
    case InitialConnectionEvent::e_AUTHN_REQUEST: {
        BSLS_ASSERT_SAFE(
            bsl::holds_alternative<bmqp_ctrlmsg::AuthenticationMessage>(
                message));
        const bmqp_ctrlmsg::AuthenticationMessage& authenticationMsg =
            bsl::get<bmqp_ctrlmsg::AuthenticationMessage>(message);

        if (oldState == InitialConnectionState::e_INITIAL) {
            BALL_LOG_INFO << "Received authentication request"
                          << " [peer: " << channel().get() << "]";

            setState(InitialConnectionState::e_AUTHENTICATING, event);

            rc = d_authenticator_p->handleAuthentication(errStream,
                                                         this,
                                                         authenticationMsg);
        }
        else {
            errStream << "Unexpected event received: " << oldState << " -> "
                      << event;
            BALL_LOG_ERROR << "#UNEXPECTED_STATE " << errStream.str()
                           << " [peer: " << channel().get() << "]";
        }
        break;
    }
    case InitialConnectionEvent::e_NEGOTIATION_MESSAGE: {
        BSLS_ASSERT_SAFE(
            bsl::holds_alternative<bmqp_ctrlmsg::NegotiationMessage>(message));
        const bmqp_ctrlmsg::NegotiationMessage& negotiationMsg =
            bsl::get<bmqp_ctrlmsg::NegotiationMessage>(message);

        if (oldState == InitialConnectionState::e_INITIAL &&
            negotiationMsg.isClientIdentityValue()) {
            BALL_LOG_INFO
                << "Received negotiation message without authentication, "
                << "performing anonymous authentication for '"
                << channel().get() << "'";

            setState(InitialConnectionState::e_ANON_AUTHENTICATING, event);

            createNegotiationContext();
            negotiationContext()->setNegotiationMessage(negotiationMsg);

            rc = handleAnonAuthentication(errStream);
        }
        else if (oldState == InitialConnectionState::e_AUTHENTICATED &&
                 negotiationMsg.isClientIdentityValue()) {
            setState(InitialConnectionState::e_NEGOTIATED, event);

            createNegotiationContext();
            negotiationContext()->setNegotiationMessage(negotiationMsg);

            rc = rc_SUCCESS;
        }
        else if (oldState == InitialConnectionState::e_NEGOTIATING_OUTBOUND &&
                 negotiationMsg.isBrokerResponseValue()) {
            setState(InitialConnectionState::e_NEGOTIATED, event);

            BSLS_ASSERT_SAFE(negotiationContext());
            negotiationContext()->setNegotiationMessage(negotiationMsg);

            rc = rc_SUCCESS;
        }
        else {
            errStream << "Unexpected event received: " << oldState << " -> "
                      << event << " [ negotiationMsg: " << negotiationMsg
                      << " ]";
            BALL_LOG_ERROR << "#UNEXPECTED_STATE " << errStream.str()
                           << " [peer: " << channel().get() << "]";
        }
        break;
    }
    case InitialConnectionEvent::e_AUTHN_SUCCESS: {
        if (oldState == InitialConnectionState::e_AUTHENTICATING) {
            setState(InitialConnectionState::e_AUTHENTICATED, event);

            // Now read Negotiation message
            rc = scheduleRead(errStream);
        }
        else if (oldState == InitialConnectionState::e_ANON_AUTHENTICATING) {
            setState(InitialConnectionState::e_NEGOTIATED, event);

            BSLS_ASSERT_SAFE(negotiationContext());
            BSLS_ASSERT_SAFE(negotiationContext()
                                 ->negotiationMessage()
                                 .isClientIdentityValue());

            rc = rc_SUCCESS;
        }
        else {
            errStream << "Unexpected event received: " << oldState << " -> "
                      << event;
            BALL_LOG_ERROR << "#UNEXPECTED_STATE " << errStream.str()
                           << " [peer: " << channel().get() << "]";
        }
        break;
    }
    case InitialConnectionEvent::e_ERROR: {
        errStream << errorDescription;
    } break;
    case InitialConnectionEvent::e_NONE: {
        errStream << "InitialConnectionContext: received e_NONE event";
        BALL_LOG_ERROR << "#UNEXPECTED_STATE " << errStream.str()
                       << " [peer: " << channel().get() << "]";
        break;
    }
    default:
        errStream << "InitialConnectionContext: "
                  << "unexpected event received: " << event;
        BALL_LOG_ERROR << "#UNEXPECTED_STATE " << errStream.str()
                       << " [peer: " << channel().get() << "]";
    }

    bsl::shared_ptr<mqbnet::Session> session;

    if (rc == rc_SUCCESS && d_state == InitialConnectionState::e_NEGOTIATED) {
        rc = d_negotiator_p->createSessionOnMsgType(errStream, &session, this);
        BALL_LOG_INFO << "Created a session with '" << channel().get() << "'";
    }

    if (rc != rc_SUCCESS) {
        setState(InitialConnectionState::e_FAILED, event);
    }

    if (hasFinalState()) {
        if (rc != rc_SUCCESS) {
            BALL_LOG_INFO << "Finished initial connection with rc = " << rc
                          << ", error = '" << errStream.str() << "'"
                          << " [peer: " << channel().get() << "]";
        }
        guard.release()->unlock();
        complete(rc, errStream.str(), session);
    }
}

// ACCESSORS
bool InitialConnectionContext::isIncoming() const
{
    return d_isIncoming;
}

const bsl::shared_ptr<NegotiationUserData>&
InitialConnectionContext::userData() const
{
    return d_userData_sp;
}

void* InitialConnectionContext::resultState() const
{
    return d_resultState_p;
}

const bsl::shared_ptr<bmqio::Channel>&
InitialConnectionContext::channel() const
{
    return d_channel_sp;
}

bmqp::EncodingType::Enum
InitialConnectionContext::authenticationEncodingType() const
{
    return d_authenticationEncodingType;
}

const bsl::shared_ptr<AuthenticationContext>&
InitialConnectionContext::authenticationContext() const
{
    return d_authenticationCtx_sp;
}

const bsl::shared_ptr<NegotiationContext>&
InitialConnectionContext::negotiationContext() const
{
    return d_negotiationCtx_sp;
}

InitialConnectionState::Enum InitialConnectionContext::state() const
{
    return d_state;
}

bool InitialConnectionContext::hasFinalState() const
{
    return d_state == InitialConnectionState::e_FAILED ||
           d_state == InitialConnectionState::e_NEGOTIATED;
}

void InitialConnectionContext::complete(
    int                                     rc,
    const bsl::string&                      error,
    const bsl::shared_ptr<mqbnet::Session>& session) const
{
    BSLS_ASSERT_SAFE(d_initialConnectionCompleteCb);

    d_initialConnectionCompleteCb(rc, error, session, channel());
}

bool InitialConnectionContext::isClosed() const
{
    return d_isClosed;
}

}  // close package namespace
}  // close enterprise namespace