Commit c7757ab0 authored by Aaryaman Sagar's avatar Aaryaman Sagar Committed by Facebook GitHub Bot

Add numSubscribers() to FanoutSender

Summary: As title

Reviewed By: Gownta

Differential Revision: D32382113

fbshipit-source-id: 88d06c821306f9452c6b74f4a49fd90631e372f8
parent 746deb0c
...@@ -74,6 +74,21 @@ bool FanoutSender<ValueType>::anySubscribers() { ...@@ -74,6 +74,21 @@ bool FanoutSender<ValueType>::anySubscribers() {
return hasSenderSet() || getSingleSender() != nullptr; return hasSenderSet() || getSingleSender() != nullptr;
} }
template <typename ValueType>
std::uint64_t FanoutSender<ValueType>::numSubscribers() const {
if (senders_.index() == 0) {
auto sender =
senders_.get(folly::tag_t<detail::ChannelBridge<ValueType>>{});
return sender ? 1 : 0;
} else if (senders_.index() == 1) {
auto senders = senders_.get(
folly::tag_t<folly::F14FastSet<detail::ChannelBridgePtr<ValueType>>>{});
return senders ? senders->size() : 0;
} else {
return 0;
}
}
template <typename ValueType> template <typename ValueType>
template <typename U> template <typename U>
void FanoutSender<ValueType>::write(U&& element) { void FanoutSender<ValueType>::write(U&& element) {
......
...@@ -63,6 +63,11 @@ class FanoutSender { ...@@ -63,6 +63,11 @@ class FanoutSender {
*/ */
bool anySubscribers(); bool anySubscribers();
/**
* Returns the number of output receivers for this fanout sender.
*/
std::uint64_t numSubscribers() const;
/** /**
* Sends the given value to all corresponding receivers. * Sends the given value to all corresponding receivers.
*/ */
......
...@@ -45,14 +45,14 @@ class PointerVariant { ...@@ -45,14 +45,14 @@ class PointerVariant {
/** /**
* Returns the zero-based index of the type that is currently held. * Returns the zero-based index of the type that is currently held.
*/ */
size_t index() { return static_cast<size_t>(storage_ & kTypeMask); } size_t index() const { return static_cast<size_t>(storage_ & kTypeMask); }
/** /**
* Returns the pointer stored in the PointerVariant, if the type matches the * Returns the pointer stored in the PointerVariant, if the type matches the
* first type. If the stored type does not match the first type, an exception * first type. If the stored type does not match the first type, an exception
* will be thrown. * will be thrown.
*/ */
inline FirstType* get(folly::tag_t<FirstType>) { inline FirstType* get(folly::tag_t<FirstType>) const {
ensureCorrectType(false /* secondType */); ensureCorrectType(false /* secondType */);
return reinterpret_cast<FirstType*>(storage_ & kPointerMask); return reinterpret_cast<FirstType*>(storage_ & kPointerMask);
} }
...@@ -62,7 +62,7 @@ class PointerVariant { ...@@ -62,7 +62,7 @@ class PointerVariant {
* second type. If the stored type does not match the second type, an * second type. If the stored type does not match the second type, an
* exception will be thrown. * exception will be thrown.
*/ */
inline SecondType* get(folly::tag_t<SecondType>) { inline SecondType* get(folly::tag_t<SecondType>) const {
ensureCorrectType(true /* secondType */); ensureCorrectType(true /* secondType */);
return reinterpret_cast<SecondType*>(storage_ & kPointerMask); return reinterpret_cast<SecondType*>(storage_ & kPointerMask);
} }
...@@ -82,7 +82,7 @@ class PointerVariant { ...@@ -82,7 +82,7 @@ class PointerVariant {
} }
private: private:
void ensureCorrectType(bool secondType) { void ensureCorrectType(bool secondType) const {
if (secondType != !!(storage_ & kTypeMask)) { if (secondType != !!(storage_ & kTypeMask)) {
throw std::runtime_error(fmt::format( throw std::runtime_error(fmt::format(
"Incorrect type specified. Given: {}, Stored: {}", "Incorrect type specified. Given: {}, Stored: {}",
......
...@@ -142,5 +142,19 @@ TEST_F(FanoutSenderFixture, ReceiversCancelled) { ...@@ -142,5 +142,19 @@ TEST_F(FanoutSenderFixture, ReceiversCancelled) {
std::move(fanoutSender).close(); std::move(fanoutSender).close();
executor_.drain(); executor_.drain();
} }
TEST_F(FanoutSenderFixture, NumSubscribers) {
auto sender = FanoutSender<int>{};
EXPECT_EQ(sender.numSubscribers(), 0);
auto receiver1 = std::make_unique<Receiver<int>>(sender.subscribe());
EXPECT_EQ(sender.numSubscribers(), 1);
auto receiver2 = std::make_unique<Receiver<int>>(sender.subscribe());
EXPECT_EQ(sender.numSubscribers(), 2);
auto receiver3 = std::make_unique<Receiver<int>>(sender.subscribe());
EXPECT_EQ(sender.numSubscribers(), 3);
}
} // namespace channels } // namespace channels
} // namespace folly } // namespace folly
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment