Middleware to handle SNS subscription confirmations

@returns
356734
function snsConfirmHandler() {
	return (req, res, next) => {
		// Handle call for SNS confirmation
		// @see http://docs.aws.amazon.com/sns/latest/dg/SendMessageToHttp.html
		if (req.headers['x-amz-sns-message-type'] === 'SubscriptionConfirmation') {
			const subscribeUrl = req.body.SubscribeURL;
			debug(`Received SubscriptionConfirmation request: ${subscribeUrl}`);

			return request({ uri: subscribeUrl }, (err, response, body) => {
				if (err) {
					return res.status(400).send({error: err.message});
				}

				// Parse the response and extract the subscription ARN
				const parser = new xml2js.Parser();
				return parser.parseString(body, function callback(err, doc) {
					if (err) {
						return res.status(400).send({error: err.message});
					}

					let subscriptionArn;
					try {
						subscriptionArn = doc.ConfirmSubscriptionResponse.ConfirmSubscriptionResult[0].SubscriptionArn[0];
						if (!subscriptionArn) {
							return res.status(400).send({ error: 'Cannot find SubscriptionArn' });
						}
						debug(`Subscription: ${subscriptionArn}`);
					} catch (e) {
						// Ignore errors related to accessing the value
						return res.status(400).send({ error: 'Cannot find SubscriptionArn' });
					}

					// Retain the subscription ARN on the request for testing
					res.subscriptionArn = subscriptionArn;
					return res.send('Subscribed');
				});
			});
		} else {
			return next();
		}
	};
}
177695
async function requestConfirmation(message, defaultAnswer = false) {
  const answerName = 'requestConfirmation';
  const answers = await inquirer.prompt([
    {
      message,
      name: answerName,
      type: 'confirm',
      default: defaultAnswer,
    },
  ]);
  return answers[answerName];
}
12267
function persistSubscriptionLocally(subscription) {
    let subscriptionJSON = JSON.stringify(subscription);
    idb.open('web-push-db', 1).then(db => {
        let tx = db.transaction(['web-push-subcription'], 'readwrite');
        tx.objectStore('web-push-subcription').put({
            id: 1,
            data: subscriptionJSON
        });
        return tx.complete;
    });
}
839585
private SnsMessage.Builder<? extends SnsMessage.Builder> unmarshallSubscriptionConfirmation(SnsJsonNode message) {
        return SnsSubscriptionConfirmation.builder(client)
                                          .withSubscribeUrl(message.getString(SUBSCRIBE_URL))
                                          .withToken(message.getString(TOKEN))
                                          .withMessage(message.getString(MESSAGE));
    }
208926
void processUnsubscribe(ServerChannel session, String clientID, List<String> topics, int messageID) {
        Log.trace("processUnsubscribe invoked, removing subscription on topics {}, for clientID <{}>", topics, clientID);

        for (String topic : topics) {
            subscriptions.removeSubscription(topic, clientID);
        }
        //ack the client
        UnsubAckMessage ackMessage = new UnsubAckMessage();
        ackMessage.setMessageID(messageID);

        Log.trace("replying with UnsubAck to MSG ID {}", messageID);
        session.write(ackMessage);
    }
411651
public void subscriptionRequestReceived(SubscriptionRequest sr) throws SubscriptionException {

        LOG.info("Subscriber -> (Hub), new subscription request received.", sr.getCallback());

        try {

            verifySubscriberRequestedSubscription(sr);

            if ("subscribe".equals(sr.getMode())) {
                LOG.info("Adding callback {} to the topic {}", sr.getCallback(), sr.getTopic());
                addCallbackToTopic(sr.getCallback(), sr.getTopic());
            } else if ("unsubscribe".equals(sr.getMode())) {
                LOG.info("Removing callback {} from the topic {}", sr.getCallback(), sr.getTopic());
                removeCallbackToTopic(sr.getCallback(), sr.getTopic());
            }

        } catch (Exception e) {
            throw new SubscriptionException(e);
        }

    }
77427
@Override
  public void doPushNotificationAnswerEvent(ServerShSession session, PushNotificationRequest request, PushNotificationAnswer answer)
      throws InternalException, IllegalDiameterStateException, RouteException, OverloadException {
    // TODO Auto-generated method stub
  }
110928
private function processGeneralSubscription()
    {
        $customerId = $this->customerSession->getCustomerId();
        if ($customerId === null) {
            $this->messageManager->addError(__('Something went wrong while saving your subscription.'));
        } else {
            try {
                $customer = $this->customerRepository->getById($customerId);
                $storeId = $this->helper->storeManager->getStore()->getId();
                $customer->setStoreId($storeId);
                $isSubscribedState = $customer->getExtensionAttributes()
                    ->getIsSubscribed();
                $isSubscribedParam = (boolean)$this->getRequest()
                    ->getParam('is_subscribed', false);
                if ($isSubscribedParam !== $isSubscribedState) {
                    $this->customerRepository->save($customer);
                    if ($isSubscribedParam) {
                        $subscribeModel = $this->subscriberFactory->create()
                            ->subscribeCustomerById($customerId);
                        $subscribeStatus = $subscribeModel->getStatus();
                        if ($subscribeStatus == Subscriber::STATUS_SUBSCRIBED) {
                            $this->messageManager->addSuccess(__('We have saved your subscription.'));
                        } else {
                            $this->messageManager->addSuccess(__('A confirmation request has been sent.'));
                        }
                    } else {
                        $this->subscriberFactory->create()
                            ->unsubscribeCustomerById($customerId);
                        $this->messageManager->addSuccess(__('We have removed your newsletter subscription.'));
                    }
                } else {
                    $this->messageManager->addSuccess(__('We have updated your subscription.'));
                }
            } catch (\Exception $e) {
                $this->messageManager->addError(__('Something went wrong while saving your subscription.'));
            }
        }
    }
454405
def verify_sns_subscription_current(self, subscription_arn, topic_name,
                                        function_arn):
        # type: (str, str, str) -> bool
        
        sns_client = self._client('sns')
        try:
            attributes = sns_client.get_subscription_attributes(
                SubscriptionArn=subscription_arn)['Attributes']
            return (
                # Splitting on ':' is safe because topic names can't have
                # a ':' char.
                attributes['TopicArn'].rsplit(':', 1)[1] == topic_name and
                attributes['Endpoint'] == function_arn
            )
        except sns_client.exceptions.NotFoundException:
            return False
148983
def service_confirmation(self, bslpdu):
        """"""
        if _debug: ProxyServiceNetworkAdapter._debug("service_confirmation %r", bslpdu)

        # build a PDU
        pdu = NPDU(bslpdu.pduData)

        # the source is from the original source, not the proxy itself
        pdu.pduSource = bslpdu.bslciAddress

        # if the proxy received a broadcast, send it upstream as a broadcast
        if isinstance(bslpdu, ProxyToServerBroadcastNPDU):
            pdu.pduDestination = LocalBroadcast()
        if _debug: ProxyServiceNetworkAdapter._debug("    - pdu: %r", pdu)

        # decode it, the nework layer needs NPDUs
        npdu = NPDU()
        npdu.decode(pdu)
        if _debug: ProxyServiceNetworkAdapter._debug("    - npdu: %r", npdu)

        # send it to the service access point for processing
        self.adapterSAP.process_npdu(self, npdu)
1213651
function newService()
    {
        $realm      = aIdentifier::DEFAULT_REALM;

        $adapter    = new IdentityCredentialDigestFile;

        // Affect Application Request/Response
        $request    = \IOC::GetIoC()->get('/HttpRequest');
        $response   = \IOC::GetIoC()->get('/HttpResponse');

        $identifier = new IdentifierHttpBasicAuth;
        $identifier
            ->setRequest($request)
            ->setResponse($response)
        ;

        $identifier->setCredentialAdapter($adapter);
        $identifier->setRealm($realm);

        $authenticator = new Authenticator($identifier, $adapter);
        return $authenticator;
    }
589146
public void setSubscriptions(List<String> subscriptions) {
        this.subscriptions.clear();
        if(subscriptions == null) return;
        for(String currentSubscription: subscriptions) {
            if (this.getNotifierName().equals(AlertService.SupportedNotifier.GUS.getName())) {
                if (currentSubscription.isEmpty() || currentSubscription.length() < 10)
                    throw new IllegalArgumentException("GUS Subscription has to contain subjectId with more than 10 characters.");
            } else if (this.getNotifierName().equals(AlertService.SupportedNotifier.EMAIL.getName())) {

                if (currentSubscription.isEmpty() || !currentSubscription.contains("@")) {
                    String errorMessage = MessageFormat.format("Email Address {0} is not allowed as @ not present.",
                            currentSubscription);
                    throw new IllegalArgumentException(errorMessage);
                }
            } else if (this.getNotifierName().equals(AlertService.SupportedNotifier.CALLBACK.getName()) && currentSubscription.isEmpty()) {
                throw new IllegalArgumentException("Callback Notifier Subscription cannot be empty.");
            }
        }
        if (subscriptions != null && !subscriptions.isEmpty()) {
            this.subscriptions.addAll(subscriptions);
        }
    }
874787
def approve_subscription(data):
    
    url = data['SubscribeURL']

    domain = urlparse(url).netloc
    pattern = getattr(
        settings,
        'BOUNCY_SUBSCRIBE_DOMAIN_REGEX',
        r"sns.[a-z0-9\-]+.amazonaws.com$"
    )
    if not re.search(pattern, domain):
        logger.error('Invalid Subscription Domain %s', url)
        return HttpResponseBadRequest('Improper Subscription Domain')

    try:
        result = urlopen(url).read()
        logger.info('Subscription Request Sent %s', url)
    except urllib.HTTPError as error:
        result = error.read()
        logger.warning('HTTP Error Creating Subscription %s', str(result))

    signals.subscription.send(
        sender='bouncy_approve_subscription',
        result=result,
        notification=data
    )

    # Return a 200 Status Code
    return HttpResponse(six.u(result))
69666
def subscribe_inbox
      # double-check locking to reduce synchronization
      if @subject_id.nil?
        # nats lazy-load needs to be outside the synchronized block
        client = nats
        @lock.synchronize do
          if @subject_id.nil?
            @subject_id = client.subscribe("#{@inbox_name}.>") do |message, _, subject|
              @handled_response = true
              handle_response(message, subject)
            end
          end
        end
      end
    end
1334852
@Override
    public void bundleChanged(BundleEvent event) {
        // We *only* react to the STOPPING event for the system bundle (bundle 0).
        if (event.getBundle().getBundleId() == 0 && event.getType() == BundleEvent.STOPPING) {
            ServiceReference<ForcedServerStop> forcedStop = bundleCtx.getServiceReference(ForcedServerStop.class);

            if (forcedStop != null) {
                // ABRUPT STOP (no quiesce stage)
                // This will make NotificationImpl#waitForCompletion stop looping
                normalServerStop.set(false);
            } else {
                // NICE / NORMAL STOP
                // Find all ServerQueisceListeners and notify them
                try {
                    quiesceListeners(bundleCtx.getServiceReferences(ServerQuiesceListener.class, null));
                } catch (InvalidSyntaxException e) {
                    // not going to happen with a null filter.
                }
            }
        }
    }
942083
public void subscribe(final Class<? extends Notification> notificationClass, final Subscription subscription) {
        if (!SUBSCRIPTION_MAP.containsKey(notificationClass)) {
            SUBSCRIPTION_MAP.put(notificationClass, new ArrayList<>());
        }
        final ArrayList<Subscription> subscriptions = SUBSCRIPTION_MAP.get(notificationClass);
        if (!subscriptions.contains(subscription)) {
            subscriptions.add(subscription);
        }
    }
341721
public int subscribe(final String subscriptionName, Object[] params) {
        // add the subscription to the HashMap with its initial value "false" which says that it is not ready
        subscriptionsAreReady.put(subscriptionName, false);
        
        // subscribe to a Meteor collection with given params
        // test error handling for invalid subscription
        return getDDP().subscribe(subscriptionName, params, new DDPListener() {
            @Override
            public void onReady(String id) {
                // mark subscription ready
                subscriptionsAreReady.replace(subscriptionName, true);

                // broadcast that subscription has been updated
                broadcastSubscriptionChanged(subscriptionName,
                        DdpMessageType.READY, null);
            }

            @Override
            public void onResult(Map<String, Object> jsonFields) {
                String msgtype = (String) jsonFields
                        .get(DdpMessageField.MSG);
                // ignore {"server_id":"GqrKrbcSeDfTYDkzQ"} web socket msgs
                if ((msgtype != null) && (msgtype.equals(DdpMessageType.ERROR))) {
                    broadcastDDPError((String) jsonFields
                            .get(DdpMessageField.ERRORMSG));
                }
            }
        });
    }