Simple Message Queue (formerly MNS):Create an HTTP subscription with the SMQ SDK for Java

Prerequisites

Before you begin, make sure that you have:

• Network connectivity between SMQ and your HTTP server so that SMQ can deliver POST requests to your endpoint

• The SMQ SDK for Java

• Apache HttpComponents and Apache Commons Codec on your classpath

How it works

When you subscribe an HTTP endpoint to an SMQ topic, SMQ delivers each published message as an HTTP POST request. Your server must:

1. Accept the incoming POST request.

2. Verify the request signature to confirm it originated from SMQ.

3. Parse the notification payload -- either XML or simplified plain text.

4. Return HTTP 204 No Content to acknowledge receipt.

The sample code in this guide implements all four steps as a standalone Java HTTP server.

Choose a notification format

SMQ supports three notification formats. Choose one based on your message content:

JSONObject object = new JSONObject(message.getMessageBodyAsRawString());
String jsonMessageData = String.valueOf(object.get("Message"));
String messageBody = new String(Base64.decodeBase64(jsonMessageData));

Notification payload reference

Each XML-format notification contains a <Notification> root element with these fields:

Verify the request signature

SMQ signs every request it sends to your endpoint. Always verify the signature before processing a notification to guard against spoofed messages.

Verification steps:

1. Read the x-mns-signing-cert-url header and Base64-decode it to get the signing certificate URL.

2. Download the X.509 certificate from that URL.

3. Build the string to sign from the request method, headers (Content-md5, Content-Type, Date), sorted x-mns-* headers, and the request URI.

4. Verify the Authorization header value against the string to sign using SHA1withRSA and the certificate's public key.

If verification fails, return HTTP 400 Bad Request and discard the notification.

Sample code

The server registers two request handlers:

• /notifications -- Parses XML-format notifications and extracts all fields.

• /simplified -- Reads SIMPLIFIED-format notifications as plain text.

Both handlers verify the request signature before processing the payload.

For the complete source, see HttpEndpointSubscription.java on GitHub.

import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.net.BindException;
import java.net.HttpURLConnection;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.URL;
import java.net.UnknownHostException;
import java.security.PublicKey;
import java.security.cert.Certificate;
import java.security.cert.CertificateFactory;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import javax.net.ssl.SSLServerSocketFactory;
import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;
import javax.xml.parsers.ParserConfigurationException;
import org.apache.commons.codec.binary.Base64;
import org.apache.http.ConnectionClosedException;
import org.apache.http.Header;
import org.apache.http.HttpConnectionFactory;
import org.apache.http.HttpEntity;
import org.apache.http.HttpEntityEnclosingRequest;
import org.apache.http.HttpException;
import org.apache.http.HttpRequest;
import org.apache.http.HttpResponse;
import org.apache.http.HttpServerConnection;
import org.apache.http.HttpStatus;
import org.apache.http.MethodNotSupportedException;
import org.apache.http.impl.DefaultBHttpServerConnection;
import org.apache.http.impl.DefaultBHttpServerConnectionFactory;
import org.apache.http.protocol.BasicHttpContext;
import org.apache.http.protocol.HttpContext;
import org.apache.http.protocol.HttpProcessor;
import org.apache.http.protocol.HttpProcessorBuilder;
import org.apache.http.protocol.HttpRequestHandler;
import org.apache.http.protocol.HttpService;
import org.apache.http.protocol.ResponseConnControl;
import org.apache.http.protocol.ResponseContent;
import org.apache.http.protocol.ResponseDate;
import org.apache.http.protocol.ResponseServer;
import org.apache.http.protocol.UriHttpRequestHandlerMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.NodeList;
import org.xml.sax.SAXException;

/**
 * HTTP endpoint server that receives SMQ topic subscription notifications.
 * Listens for POST requests on /notifications (XML format) and /simplified (plain text).
 * Verifies each request's signature before processing.
 */
public class HttpEndpointSubscription {
    public static Logger logger = LoggerFactory.getLogger(HttpEndpointSubscription.class);
    public static Thread t;
    private int port;

    /**
     * Generate an HTTP endpoint URL from the local host IP address, using port 80.
     *
     * @return HTTP endpoint URL
     */
    public static String genEndpointLocal() {
        return HttpEndpointSubscription.genEndpointLocal(80);
    }

    /**
     * Generate an HTTP endpoint URL from the local host IP address.
     *
     * @param port HTTP server port
     * @return HTTP endpoint URL
     */
    public static String genEndpointLocal(int port) {
        try {
            InetAddress addr = InetAddress.getLocalHost();
            String ip = addr.getHostAddress().toString();
            return "http://" + ip + ":" + port;
        } catch (UnknownHostException e) {
            e.printStackTrace();
            logger.warn("get local host fail," + e.getMessage());
            return "http://127.0.0.1:" + port;
        }

    }

    /**
     * Create an HttpEndpointSubscription on the specified port.
     *
     * @param port HTTP server port
     */
    public HttpEndpointSubscription(int port) {
        init(port);
    }

    /**
     * Create an HttpEndpointSubscription on port 80 (default).
     */
    public HttpEndpointSubscription() {
        init(80);
    }

    private void init(int port) {
        this.port = port;
        t = null;
    }

    /**
     * Start the HTTP server.
     *
     * @throws Exception if the port is already in use or the server fails to start
     */
    public void start() throws Exception {
        // Check whether the port is already in use
        try {
            new Socket(InetAddress.getLocalHost(), this.port);
            System.out.println("port is used!");
            logger.error("port already in use, http server start failed");
            throw new BindException("port already in use");
        } catch (IOException e) {
            // Port is available
        }

        // Set up the HTTP protocol processor
        HttpProcessor httpproc = HttpProcessorBuilder.create()
            .add(new ResponseDate())
            .add(new ResponseServer("MNS-Endpoint/1.1"))
            .add(new ResponseContent())
            .add(new ResponseConnControl()).build();

        // Register request handlers for each notification format
        UriHttpRequestHandlerMapper reqistry = new UriHttpRequestHandlerMapper();
        reqistry.register("/notifications", new NSHandler());
        reqistry.register("/simplified", new SimplifiedNSHandler());

        // Set up the HTTP service
        HttpService httpService = new HttpService(httpproc, reqistry);

        // Start the listener thread
        t = new RequestListenerThread(port, httpService, null);
        t.setDaemon(false);
        t.start();
    }

    /**
     * Stop the HTTP server.
     */
    public void stop() {
        if (t != null) {
            t.interrupt();
            try {
                t.join(10 * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("endpoint stop");
    }

    /**
     * Verify that the request originated from SMQ by checking its signature
     * against the signing certificate.
     *
     * @param method  HTTP method (POST)
     * @param uri     Request URI
     * @param headers Request headers
     * @param cert    Signing certificate URL
     * @return true if the signature is valid
     */
    private Boolean authenticate(String method, String uri, Map<String, String> headers, String cert) {
        String str2sign = getSignStr(method, uri, headers);
        String signature = headers.get("Authorization");
        byte[] decodedSign = Base64.decodeBase64(signature);
        // Download the X.509 certificate and verify the signature
        try {
            URL url = new URL(cert);
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            DataInputStream in = new DataInputStream(conn.getInputStream());
            CertificateFactory cf = CertificateFactory.getInstance("X.509");

            Certificate c = cf.generateCertificate(in);
            PublicKey pk = c.getPublicKey();

            java.security.Signature signetcheck = java.security.Signature.getInstance("SHA1withRSA");
            signetcheck.initVerify(pk);
            signetcheck.update(str2sign.getBytes());
            Boolean res = signetcheck.verify(decodedSign);
            return res;
        } catch (Exception e) {
            e.printStackTrace();
            logger.warn("authenticate fail, " + e.getMessage());
            return false;
        }
    }

    /**
     * Build the string to sign from request components.
     * Format: METHOD\nContent-md5\nContent-Type\nDate\n[sorted x-mns-* headers]\nURI
     *
     * @param method  HTTP method
     * @param uri     Request URI
     * @param headers Request headers
     * @return String to sign
     */
    private String getSignStr(String method, String uri, Map<String, String> headers) {
        StringBuilder sb = new StringBuilder();
        sb.append(method);
        sb.append("\n");
        sb.append(safeGetHeader(headers, "Content-md5"));
        sb.append("\n");
        sb.append(safeGetHeader(headers, "Content-Type"));
        sb.append("\n");
        sb.append(safeGetHeader(headers, "Date"));
        sb.append("\n");

        List<String> tmp = new ArrayList<String>();
        for (Map.Entry<String, String> entry : headers.entrySet()) {
            if (entry.getKey().startsWith("x-mns-")) {
                tmp.add(entry.getKey() + ":" + entry.getValue());
            }
        }
        Collections.sort(tmp);

        for (String kv : tmp) {
            sb.append(kv);
            sb.append("\n");
        }

        sb.append(uri);
        return sb.toString();
    }

    private String safeGetHeader(Map<String, String> headers, String name) {
        if (headers.containsKey(name)) {
            return headers.get(name);
        } else {
            return "";
        }
    }

    /**
     * Handles SIMPLIFIED-format notifications on /simplified.
     * Reads the raw request body as plain text after verifying the signature.
     */
    public class SimplifiedNSHandler implements HttpRequestHandler {
        @Override
        public void handle(
            final HttpRequest request,
            final HttpResponse response,
            final HttpContext context) throws HttpException, IOException {

            String method = request.getRequestLine().getMethod().toUpperCase(Locale.ENGLISH);

            if (!method.equals("GET") && !method.equals("HEAD") && !method.equals("POST")) {
                throw new MethodNotSupportedException(method + " method not supported");
            }

            Header[] headers = request.getAllHeaders();
            Map<String, String> hm = new HashMap<String, String>();
            for (Header h : headers) {
                System.out.println(h.getName() + ":" + h.getValue());
                hm.put(h.getName(), h.getValue());
            }

            String target = request.getRequestLine().getUri();
            System.out.println(target);

            if (request instanceof HttpEntityEnclosingRequest) {
                HttpEntity entity = ((HttpEntityEnclosingRequest) request).getEntity();

                // Verify the request signature
                Header certHeader = request.getFirstHeader("x-mns-signing-cert-url");
                if (certHeader == null) {
                    System.out.println("SigningCerURL Header not found");
                    response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                    return;
                }

                String cert = certHeader.getValue();
                if (cert.isEmpty()) {
                    System.out.println("SigningCertURL empty");
                    response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                    return;
                }
                cert = new String(Base64.decodeBase64(cert));
                System.out.println("SigningCertURL:\t" + cert);
                logger.debug("SigningCertURL:\t" + cert);

                if (!authenticate(method, target, hm, cert)) {
                    System.out.println("authenticate fail");
                    logger.warn("authenticate fail");
                    response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                    return;
                }

                // Read the simplified notification body
                InputStream is = entity.getContent();
                BufferedReader reader = new BufferedReader(new InputStreamReader(is));
                StringBuffer buffer = new StringBuffer();
                String line = "";
                while ((line = reader.readLine()) != null) {
                    buffer.append(line);
                }
                String content = buffer.toString();

                System.out.println("Simplified Notification: \n" + content);
            }

            response.setStatusCode(HttpStatus.SC_NO_CONTENT);
        }
    }

    /**
     * Handles XML-format notifications on /notifications.
     * Parses the XML payload and extracts all notification fields after verifying the signature.
     */
    public class NSHandler implements HttpRequestHandler {
        public Logger logger = LoggerFactory.getLogger(HttpRequestHandler.class);

        public NSHandler() {
            super();
        }

        private String safeGetElementContent(Element element, String tag) {
            NodeList nl = element.getElementsByTagName(tag);
            if (nl != null && nl.getLength() > 0) {
                return nl.item(0).getTextContent();
            } else {
                logger.warn("get " + tag + " from xml fail");
                return "";
            }
        }

        /**
         * Parse notification fields from the XML element.
         *
         * @param notify XML Notification element
         */
        private void parserContent(Element notify) {
            try {
                String topicOwner = safeGetElementContent(notify, "TopicOwner");
                System.out.println("TopicOwner:\t" + topicOwner);
                logger.debug("TopicOwner:\t" + topicOwner);

                String topicName = safeGetElementContent(notify, "TopicName");
                System.out.println("TopicName:\t" + topicName);
                logger.debug("TopicName:\t" + topicName);

                String subscriber = safeGetElementContent(notify, "Subscriber");
                System.out.println("Subscriber:\t" + subscriber);
                logger.debug("Subscriber:\t" + subscriber);

                String subscriptionName = safeGetElementContent(notify, "SubscriptionName");
                System.out.println("SubscriptionName:\t" + subscriptionName);
                logger.debug("SubscriptionName:\t" + subscriptionName);

                String msgid = safeGetElementContent(notify, "MessageId");
                System.out.println("MessageId:\t" + msgid);
                logger.debug("MessageId:\t" + msgid);

                // Base64-decode the message content
                String msg = safeGetElementContent(notify, "Message");
                System.out.println("Message:\t" + new String(Base64.decodeBase64(msg)));
                logger.debug("Message:\t" + new String(Base64.decodeBase64(msg)));

                String msgMD5 = safeGetElementContent(notify, "MessageMD5");
                System.out.println("MessageMD5:\t" + msgMD5);
                logger.debug("MessageMD5:\t" + msgMD5);

                String msgPublishTime = safeGetElementContent(notify, "PublishTime");
                Date d = new Date(Long.parseLong(msgPublishTime));
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                String strdate = sdf.format(d);
                System.out.println("PublishTime:\t" + strdate);
                logger.debug("MessagePublishTime:\t" + strdate);

                String msgTag = safeGetElementContent(notify, "MessageTag");
                if (!msgTag.equals("")) {
                    System.out.println("MessageTag:\t" + msgTag);
                    logger.debug("MessageTag:\t" + msgTag);
                }

            } catch (Exception e) {
                System.out.println(e.getMessage());
                e.printStackTrace();
                logger.warn(e.getMessage());
            }

        }

        @Override
        public void handle(
            final HttpRequest request,
            final HttpResponse response,
            final HttpContext context) throws HttpException, IOException {

            String method = request.getRequestLine().getMethod().toUpperCase(Locale.ENGLISH);

            if (!method.equals("GET") && !method.equals("HEAD") && !method.equals("POST")) {
                throw new MethodNotSupportedException(method + " method not supported");
            }

            Header[] headers = request.getAllHeaders();
            Map<String, String> hm = new HashMap<String, String>();
            for (Header h : headers) {
                System.out.println(h.getName() + ":" + h.getValue());
                hm.put(h.getName(), h.getValue());
            }

            String target = request.getRequestLine().getUri();
            System.out.println(target);

            if (request instanceof HttpEntityEnclosingRequest) {
                HttpEntity entity = ((HttpEntityEnclosingRequest) request).getEntity();

                // Parse the XML notification payload
                InputStream content = entity.getContent();
                DocumentBuilderFactory dbf = DocumentBuilderFactory.newInstance();
                Element notify = null;
                try {
                    DocumentBuilder db = dbf.newDocumentBuilder();
                    Document document = db.parse(content);
                    NodeList nl = document.getElementsByTagName("Notification");
                    if (nl == null || nl.getLength() == 0) {
                        System.out.println("xml tag error");
                        logger.warn("xml tag error");
                        response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                        return;
                    }
                    notify = (Element) nl.item(0);
                } catch (ParserConfigurationException e) {
                    e.printStackTrace();
                    logger.warn("xml parser fail!  " + e.getMessage());
                    response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                    return;
                } catch (SAXException e) {
                    e.printStackTrace();
                    logger.warn("xml parser fail!  " + e.getMessage());
                    response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                    return;
                }

                // Verify the request signature
                Header certHeader = request.getFirstHeader("x-mns-signing-cert-url");
                if (certHeader == null) {
                    System.out.println("SigningCerURL Header not found");
                    response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                    return;
                }

                String cert = certHeader.getValue();
                if (cert.isEmpty()) {
                    System.out.println("SigningCertURL empty");
                    response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                    return;
                }
                cert = new String(Base64.decodeBase64(cert));
                System.out.println("SigningCertURL:\t" + cert);
                logger.debug("SigningCertURL:\t" + cert);

                if (!authenticate(method, target, hm, cert)) {
                    System.out.println("authenticate fail");
                    logger.warn("authenticate fail");
                    response.setStatusCode(HttpStatus.SC_BAD_REQUEST);
                    return;
                }
                parserContent(notify);

            }

            response.setStatusCode(HttpStatus.SC_NO_CONTENT);
        }

    }

    /**
     * Listener thread that accepts incoming HTTP connections and dispatches
     * each to a worker thread for processing.
     */
    public class RequestListenerThread extends Thread {

        private final HttpConnectionFactory<DefaultBHttpServerConnection> connFactory;
        private final ServerSocket serversocket;
        private final HttpService httpService;

        public RequestListenerThread(
            final int port,
            final HttpService httpService,
            final SSLServerSocketFactory sf) throws IOException {
            this.connFactory = DefaultBHttpServerConnectionFactory.INSTANCE;
            this.serversocket = sf != null ?  sf.createServerSocket(port) : new ServerSocket(port);
            this.httpService = httpService;
        }

        @Override
        public void run() {
            System.out.println("Listening on port " + this.serversocket.getLocalPort());
            Thread t = null;
            while (!Thread.interrupted()) {
                try {
                    // Accept incoming connections
                    Socket socket = this.serversocket.accept();
                    System.out.println("Incoming connection from " + socket.getInetAddress());
                    HttpServerConnection conn = this.connFactory.createConnection(socket);

                    // Dispatch to a worker thread
                    t = new WorkerThread(this.httpService, conn);
                    t.setDaemon(true);
                    t.start();
                } catch (IOException e) {
                    System.err.println("Endpoint http server stop or IO error: "
                        + e.getMessage());
                    try {
                        if (t != null) {
                            t.join(5 * 1000);
                        }
                    } catch (InterruptedException e1) {
                        // Ignored
                    }
                    break;
                }
            }
        }

        @Override
        public void interrupt() {
            super.interrupt();
            try {
                this.serversocket.close();
            } catch (IOException e) {
                // Ignored
            }
        }
    }

    /**
     * Worker thread that processes a single HTTP connection.
     */
    public class WorkerThread extends Thread {

        private final HttpService httpservice;
        private final HttpServerConnection conn;

        public WorkerThread(
            final HttpService httpservice,
            final HttpServerConnection conn) {
            super();
            this.httpservice = httpservice;
            this.conn = conn;
        }

        @Override
        public void run() {
            System.out.println("New connection thread");
            HttpContext context = new BasicHttpContext(null);
            try {
                while (!Thread.interrupted() && this.conn.isOpen()) {
                    this.httpservice.handleRequest(this.conn, context);
                }
            } catch (ConnectionClosedException ex) {
                System.err.println("Client closed connection");
            } catch (IOException ex) {
                System.err.println("I/O error: " + ex.getMessage());
            } catch (HttpException ex) {
                System.err.println("Unrecoverable HTTP protocol violation: " + ex.getMessage());
            } finally {
                try {
                    this.conn.shutdown();
                } catch (IOException ignore) {
                }
            }
        }

    }

    /**
     * Start the HTTP endpoint server on port 8080.
     *
     * @param args command-line arguments (not used)
     */
    public static void main(String[] args) {
        int port = 8080;
        HttpEndpointSubscription httpEndpointSubscription = null;
        try {
            httpEndpointSubscription = new HttpEndpointSubscription(port);
            httpEndpointSubscription.start();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (httpEndpointSubscription != null) {
                httpEndpointSubscription.stop();
            }
        }
    }

}

# Compile and run with required dependencies on the classpath
javac -cp ".:lib/*" HttpEndpointSubscription.java
java -cp ".:lib/*" HttpEndpointSubscription