C++以http接口推送json流和图片流

一、C++推送json流

C++代码：

//
// a single-threaded, multi client(using select), debug webserver - streaming out mjpg.
//  on win, _WIN32 has to be defined, must link against ws2_32.lib (socks on linux are for free)
//

//
// socket related abstractions:
//
#ifdef _WIN32  
    #include <winsock.h>
    #include <windows.h>
    #include <time.h>
    #define PORT        unsigned long
    #define ADDRPOINTER   int*
    struct _INIT_W32DATA
    {
       WSADATA w;
       _INIT_W32DATA() { WSAStartup( MAKEWORD( 2, 1 ), &w ); }
    } _init_once;
#else       /* ! win32 */
    #include <unistd.h>
    #include <sys/time.h>
    #include <sys/types.h>
    #include <sys/socket.h>
    #include <netdb.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #define PORT        unsigned short
    #define SOCKET    int
    #define HOSTENT  struct hostent
    #define SOCKADDR    struct sockaddr
    #define SOCKADDR_IN  struct sockaddr_in
    #define ADDRPOINTER  unsigned int*
    #define INVALID_SOCKET -1
    #define SOCKET_ERROR   -1
#endif /* _WIN32 */


#include<signal.h>

#include <iostream>
using std::cerr;
using std::endl;

#include "opencv2/opencv.hpp"
using namespace cv;

static int close_socket(SOCKET s) {
    std::cout << "close_socket function "<< endl;
    int close_output = ::shutdown(s, 1); // 0 close input, 1 close output, 2 close both
    char *buf = (char *)calloc(1024, sizeof(char));
    ::recv(s, buf, 1024, 0);
    free(buf);
    int close_input = ::shutdown(s, 0);
    int result = close(s);
    std::cerr << "Close socket: out = " << close_output << ", in = " << close_input << " \n";
    return result;
}


//struct mat_cv : cv::Mat { int a[0]; };

// void send_mjpeg(mat_cv* mat, int port, int timeout, int quality)
// {
//     try {
//         std::lock_guard<std::mutex> lock(mtx_mjpeg);
//         static MJPG_sender wri(port, timeout, quality);
//         //cv::Mat mat = cv::cvarrToMat(ipl);
//         wri.write(*(cv::Mat*)mat);
//         std::cout << " MJPEG-stream sent. \n";
//     }
//     catch (...) {
//         cerr << " Error in send_mjpeg() function \n";
//     }
// }


// =====================================================================================================================================================
class JSON_sender
{
    SOCKET sock;
    SOCKET maxfd;
    fd_set master;
    int timeout; // master sock timeout, shutdown after timeout usec.
    int close_all_sockets;

    int _write(int sock, char const*const s, int len)
    {
        if (len < 1) { len = strlen(s); }
        return ::send(sock, s, len, 0);
    }

public:

    JSON_sender(int port = 0, int _timeout = 400000)
        : sock(INVALID_SOCKET)
        , timeout(_timeout)
    {
        close_all_sockets = 0;
        FD_ZERO(&master);
        if (port)
            open(port);
    }

    ~JSON_sender()
    {
        std::cout << "exit JSON_sender "<< std::endl;
        close_all();
        release();
    }

    bool release()
    {
        if (sock != INVALID_SOCKET)
            ::shutdown(sock, 2);
        sock = (INVALID_SOCKET);
        return false;
    }

    void close_all()
    {
        close_all_sockets = 1;
        write("\n]");   // close JSON array
    }

    bool open(int port)
    {
        sock = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

        SOCKADDR_IN address;
        address.sin_addr.s_addr = INADDR_ANY;
        address.sin_family = AF_INET;
        address.sin_port = htons(port);    // ::htons(port);
        int reuse = 1;
        if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (const char*)&reuse, sizeof(reuse)) < 0)
            cerr << "setsockopt(SO_REUSEADDR) failed" << endl;

        // Non-blocking sockets
        // Windows: ioctlsocket() and FIONBIO
        // Linux: fcntl() and O_NONBLOCK
#ifdef WIN32
        unsigned long i_mode = 1;
        int result = ioctlsocket(sock, FIONBIO, &i_mode);
        if (result != NO_ERROR) {
            std::cerr << "ioctlsocket(FIONBIO) failed with error: " << result << std::endl;
            signal(SIGPIPE, SIG_IGN);
        }
#else // WIN32
        ;
        // int flags = fcntl(sock, F_GETFL, 0);
        // fcntl(sock, F_SETFL, flags | O_NONBLOCK);
#endif // WIN32

#ifdef SO_REUSEPORT
        if (setsockopt(sock, SOL_SOCKET, SO_REUSEPORT, (const char*)&reuse, sizeof(reuse)) < 0){
            //signal(SIGPIPE, SIG_IGN);
            cerr << "setsockopt(SO_REUSEPORT) failed" << endl;
        }
#endif
        if (::bind(sock, (SOCKADDR*)&address, sizeof(SOCKADDR_IN)) == SOCKET_ERROR)
        {
            cerr << "error JSON_sender: couldn't bind sock " << sock << " to port " << port << "!" << endl;
            //signal(SIGPIPE, SIG_IGN);
            return release();
        }
        if (::listen(sock, 10) == SOCKET_ERROR)
        {
            cerr << "error JSON_sender: couldn't listen on sock " << sock << " on port " << port << " !" << endl;
            //signal(SIGPIPE, SIG_IGN);
            return release();
        }
        FD_ZERO(&master);
        FD_SET(sock, &master);
        maxfd = sock;
        return true;
    }

    bool isOpened()
    {
        return sock != INVALID_SOCKET;
    }

    bool write(char const* outputbuf)
    {
        fd_set rread = master;
        struct timeval select_timeout = { 0, 0 };
        struct timeval socket_timeout = { 0, timeout };
        if (::select(maxfd + 1, &rread, NULL, NULL, &select_timeout) <= 0){
            signal(SIGPIPE, SIG_IGN);
            return true; // nothing broken, there's just noone listening
        }
        
        int outlen = static_cast<int>(strlen(outputbuf));

#ifdef _WIN32
        for (unsigned i = 0; i<rread.fd_count; i++)
        {
            int addrlen = sizeof(SOCKADDR);
            SOCKET s = rread.fd_array[i];    // fd_set on win is an array, while ...
#else
        for (int s = 0; s <= maxfd; s++)
        {
            socklen_t addrlen = sizeof(SOCKADDR);
            if (!FD_ISSET(s, &rread))      // ... on linux it's a bitmask ;)
                continue;
#endif
            if (s == sock) // request on master socket, accept and send main header.
            {
                SOCKADDR_IN address = { 0 };
                SOCKET      client = ::accept(sock, (SOCKADDR*)&address, &addrlen);
                if (client == SOCKET_ERROR)
                {
                    cerr << "error JSON_sender: couldn't accept connection on sock " << sock << " !" << endl;
                    //signal(SIGPIPE, SIG_IGN);
                    return false;
                }
                if (setsockopt(client, SOL_SOCKET, SO_RCVTIMEO, (char *)&socket_timeout, sizeof(socket_timeout)) < 0) {
                    //signal(SIGPIPE, SIG_IGN);
                    cerr << "error JSON_sender: SO_RCVTIMEO setsockopt failed\n";
                }
                if (setsockopt(client, SOL_SOCKET, SO_SNDTIMEO, (char *)&socket_timeout, sizeof(socket_timeout)) < 0) {
                    //signal(SIGPIPE, SIG_IGN);
                    cerr << "error JSON_sender: SO_SNDTIMEO setsockopt failed\n";
                }
                maxfd = (maxfd>client ? maxfd : client);
                FD_SET(client, &master);
                _write(client, "HTTP/1.0 200 OK\r\n", 0);
                _write(client,
                    "Server: Mozarella/2.2\r\n"
                    "Accept-Range: bytes\r\n"
                    "Connection: close\r\n"
                    "Max-Age: 0\r\n"
                    "Expires: 0\r\n"
                    "Cache-Control: no-cache, private\r\n"
                    "Pragma: no-cache\r\n"
                    "Content-Type: application/json\r\n"
                    //"Content-Type: multipart/x-mixed-replace; boundary=boundary\r\n"
                    "\r\n", 0);
                _write(client, "[\n", 0);   // open JSON array
                int n = _write(client, outputbuf, outlen);
                cerr << "JSON_sender: new client " << client << endl;
            }
            else // existing client, just stream pix
            {
                //char head[400];
                // application/x-resource+json or application/x-collection+json -  when you are representing REST resources and collections
                // application/json or text/json or text/javascript or text/plain.
                // https://stackoverflow.com/questions/477816/what-is-the-correct-json-content-type
                //sprintf(head, "\r\nContent-Length: %zu\r\n\r\n", outlen);
                //sprintf(head, "--boundary\r\nContent-Type: application/json\r\nContent-Length: %zu\r\n\r\n", outlen);
                //_write(s, head, 0);
                if (!close_all_sockets) _write(s, ", \n", 0);
                int n = _write(s, outputbuf, outlen);
                if (n < (int)outlen)
                {
                    cerr << "JSON_sender: kill client " << s << endl;
                    //signal(SIGPIPE, SIG_IGN);
                    //close_socket(s);
                    //::shutdown(s, 2);
                    FD_CLR(s, &master);
                }

                if (close_all_sockets) {
                    int result = close_socket(s);
                    cerr << "JSON_sender: close clinet: " << result << " \n";
                    //signal(SIGPIPE, SIG_IGN);
                    continue;
                }
            }
        }
        if (close_all_sockets) {
            int result = close_socket(sock);
            cerr << "JSON_sender: close acceptor: " << result << " \n\n";
        }
        return true;
        }
};
// ----------------------------------------


int main()
{
    static std::unique_ptr<JSON_sender> js_ptr;
    //static std::mutex mtx;


    while(1)
    {
        try{
            if(!js_ptr) js_ptr.reset(new JSON_sender(22221, 400000));
            
            js_ptr->write("hello world");
            std::cout << " JSON-stream sent. \n";
        }catch (...) {
            std::cout << " Error in send_json() function \n";
        }   
        

    }
    std::cout << "finish" << std::endl;
    return 0;
}

打开浏览器，输入ip地址:22221，就可以看到推送的json流了

二、C++推送图片流

C++代码：

//
// a single-threaded, multi client(using select), debug webserver - streaming out mjpg.
//  on win, _WIN32 has to be defined, must link against ws2_32.lib (socks on linux are for free)
//

//
// socket related abstractions:
//
#ifdef _WIN32  
    #include <winsock.h>
    #include <windows.h>
    #include <time.h>
    #define PORT        unsigned long
    #define ADDRPOINTER   int*
    struct _INIT_W32DATA
    {
       WSADATA w;
       _INIT_W32DATA() { WSAStartup( MAKEWORD( 2, 1 ), &w ); }
    } _init_once;
#else       /* ! win32 */
    #include <unistd.h>
    #include <sys/time.h>
    #include <sys/types.h>
    #include <sys/socket.h>
    #include <netdb.h>
    #include <netinet/in.h>
    #include <arpa/inet.h>
    #define PORT        unsigned short
    #define SOCKET    int
    #define HOSTENT  struct hostent
    #define SOCKADDR    struct sockaddr
    #define SOCKADDR_IN  struct sockaddr_in
    #define ADDRPOINTER  unsigned int*
    #define INVALID_SOCKET -1
    #define SOCKET_ERROR   -1
#endif /* _WIN32 */


#include<signal.h>

#include <iostream>
using std::cerr;
using std::endl;

#include "opencv2/opencv.hpp"
using namespace cv;

// static int close_socket(SOCKET s) {
    // int close_output = ::shutdown(s, 1); // 0 close input, 1 close output, 2 close both
    // char *buf = (char *)calloc(1024, sizeof(char));
    // ::recv(s, buf, 1024, 0);
    // free(buf);
    // int close_input = ::shutdown(s, 0);
    // int result = close(s);
    // std::cerr << "Close socket: out = " << close_output << ", in = " << close_input << " \n";
    // return result;
// }


class MJPG_sender
{
    SOCKET sock;
    SOCKET maxfd;
    fd_set master;
    int timeout; // master sock timeout, shutdown after timeout usec.
    int quality; // jpeg compression [1..100]
    int close_all_sockets;

    int _write(int sock, char const*const s, int len)
    {
        if (len < 1) { len = strlen(s); }
        return ::send(sock, s, len, 0);
    }

public:

    MJPG_sender(int port = 0, int _timeout = 400000, int _quality = 30)
        : sock(INVALID_SOCKET)
        , timeout(_timeout)
        , quality(_quality)
    {
        close_all_sockets = 0;
        FD_ZERO(&master);
        if (port)
            open(port);
    }

    ~MJPG_sender()
    {
        //close_all();
        release();
    }

    bool release()
    {
        if (sock != INVALID_SOCKET)
            ::shutdown(sock, 2);
        sock = (INVALID_SOCKET);
        return false;
    }

    void close_all()
    {
        close_all_sockets = 1;
        cv::Mat tmp(cv::Size(10, 10), CV_8UC3);
        write(tmp);
    }

    bool open(int port)
    {
        sock = ::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);

        SOCKADDR_IN address;
        address.sin_addr.s_addr = INADDR_ANY;
        address.sin_family = AF_INET;
        address.sin_port = htons(port);    // ::htons(port);
        int reuse = 1;
        if (setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, (const char*)&reuse, sizeof(reuse)) < 0)
            cerr << "setsockopt(SO_REUSEADDR) failed" << endl;
            signal(SIGPIPE, SIG_IGN);

        // Non-blocking sockets
        // Windows: ioctlsocket() and FIONBIO
        // Linux: fcntl() and O_NONBLOCK
#ifdef WIN32
        unsigned long i_mode = 1;
        int result = ioctlsocket(sock, FIONBIO, &i_mode);
        if (result != NO_ERROR) {
            std::cerr << "ioctlsocket(FIONBIO) failed with error: " << result << std::endl;
            signal(SIGPIPE, SIG_IGN);
        }
#else // WIN32
        ;
        // int flags = fcntl(sock, F_GETFL, 0);
        // fcntl(sock, F_SETFL, flags | O_NONBLOCK);
#endif // WIN32

#ifdef SO_REUSEPORT
        if (setsockopt(sock, SOL_SOCKET, SO_REUSEPORT, (const char*)&reuse, sizeof(reuse)) < 0)
            cerr << "setsockopt(SO_REUSEPORT) failed" << endl;
            signal(SIGPIPE, SIG_IGN);
#endif
        if (::bind(sock, (SOCKADDR*)&address, sizeof(SOCKADDR_IN)) == SOCKET_ERROR)
        {
            cerr << "error MJPG_sender: couldn't bind sock " << sock << " to port " << port << "!" << endl;
            signal(SIGPIPE, SIG_IGN);
            return release();
        }
        if (::listen(sock, 10) == SOCKET_ERROR)
        {
            cerr << "error MJPG_sender: couldn't listen on sock " << sock << " on port " << port << " !" << endl;
            signal(SIGPIPE, SIG_IGN);
            return release();
        }
        FD_ZERO(&master);
        FD_SET(sock, &master);
        maxfd = sock;
        return true;
    }

    bool isOpened()
    {
        return sock != INVALID_SOCKET;
    }

    bool write(const Mat & frame)
    {
        fd_set rread = master;
        struct timeval select_timeout = { 0, 0 };
        struct timeval socket_timeout = { 0, timeout };
        struct timeval to = {0,timeout};
        //SOCKET maxfd = sock+1;
        if (::select(maxfd+1, &rread, NULL, NULL, &to) <= 0){
            std::cout << "111111111111"<<std::endl;
            signal(SIGPIPE, SIG_IGN);
            return true; // nothing broken, there's just noone listening
        }
            


        std::vector<uchar> outbuf;
        std::vector<int> params;
        params.push_back(IMWRITE_JPEG_QUALITY);
        params.push_back(quality);
        cv::imencode(".jpg", frame, outbuf, params);  //REMOVED FOR COMPATIBILITY
        // https://docs.opencv.org/3.4/d4/da8/group__imgcodecs.html#ga292d81be8d76901bff7988d18d2b42ac
        //std::cerr << "cv::imencode call disabled!" << std::endl;
        int outlen = static_cast<int>(outbuf.size());

#ifdef _WIN32
        for (unsigned i = 0; i<rread.fd_count; i++)
        {
            //int addrlen = sizeof(SOCKADDR);
            SOCKET s = rread.fd_array[i];    // fd_set on win is an array, while ...
#else
        for (int s = 0; s <= maxfd+10; s++)
        //for(int s= 0;s<10000;s++)
        {
            socklen_t addrlen = sizeof(SOCKADDR);
            if (!FD_ISSET(s, &rread))      // ... on linux it's a bitmask ;)
                continue;
#endif
            if (s == sock) // request on master socket, accept and send main header.
            {
                std::cout << "2222222222222"<<std::endl;
                SOCKADDR_IN address = { 0 };
                SOCKET      client = ::accept(sock, (SOCKADDR*)&address, &addrlen);
                if (client == SOCKET_ERROR)
                {
                    cerr << "error MJPG_sender: couldn't accept connection on sock " << sock << " !" << endl;
                    signal(SIGPIPE, SIG_IGN);
                    return false;
                }
                // if (setsockopt(client, SOL_SOCKET, SO_RCVTIMEO, (char *)&socket_timeout, sizeof(socket_timeout)) < 0) {
                    // cerr << "error MJPG_sender: SO_RCVTIMEO setsockopt failed\n";
                // }
                // if (setsockopt(client, SOL_SOCKET, SO_SNDTIMEO, (char *)&socket_timeout, sizeof(socket_timeout)) < 0) {
                    // cerr << "error MJPG_sender: SO_SNDTIMEO setsockopt failed\n";
                // }
                maxfd = (maxfd>client ? maxfd : client);
                FD_SET(client, &master);
                _write(client, "HTTP/1.0 200 OK\r\n", 0);
                _write(client,
                    "Server: Mozarella/2.2\r\n"
                    "Accept-Range: bytes\r\n"
                    "Connection: close\r\n"
                    "Max-Age: 0\r\n"
                    "Expires: 0\r\n"
                    "Cache-Control: no-cache, private\r\n"
                    "Pragma: no-cache\r\n"
                    "Content-Type: multipart/x-mixed-replace; boundary=mjpegstream\r\n"
                    "\r\n", 0);
                cerr << "MJPG_sender: new client " << client << endl;
            }
            else // existing client, just stream pix
            {
                std::cout << "333333333333333333"<<std::endl;
                // if (close_all_sockets) {
                    // int result = close_socket(s);
                    // cerr << "MJPG_sender: close clinet: " << result << " \n";
                    // continue;
                // }

                char head[400];
                sprintf(head, "--mjpegstream\r\nContent-Type: image/jpeg\r\nContent-Length: %zu\r\n\r\n", outlen);
                _write(s, head, 0);
                int n = _write(s, (char*)(&outbuf[0]), outlen);
                cerr << "known client: " << s << ", sent = " << n << ", must be sent outlen = " << outlen << endl;
                if (n < (int)outlen)
                {
                    cerr << "MJPG_sender: kill client " << s << endl;
                    signal(SIGPIPE, SIG_IGN);
                    //::shutdown(s, 2);
                    //close_socket(s);
                    FD_CLR(s, &master);
                }
            }
        }
        // if (close_all_sockets) {
            // int result = close_socket(sock);
            // cerr << "MJPG_sender: close acceptor: " << result << " \n\n";
        // }
        return true;
    }
};
// ----------------------------------------



//struct mat_cv : cv::Mat { int a[0]; };

// void send_mjpeg(mat_cv* mat, int port, int timeout, int quality)
// {
//     try {
//         std::lock_guard<std::mutex> lock(mtx_mjpeg);
//         static MJPG_sender wri(port, timeout, quality);
//         //cv::Mat mat = cv::cvarrToMat(ipl);
//         wri.write(*(cv::Mat*)mat);
//         std::cout << " MJPEG-stream sent. \n";
//     }
//     catch (...) {
//         cerr << " Error in send_mjpeg() function \n";
//     }
// }


int main()
{
    //static std::mutex mtx_mjpeg;
    VideoCapture cap;
    bool ok = cap.open("rtsp://admin:123456@192.168.10.11:554/h264/ch01/main/av_stream");
    if ( ! ok ) 
    {
        printf("no cam found ;(.\n");
        return 1;
    }

    //MJPGWriter wri(22222); // http://your-server:7777
    
    //while( cap.isOpened() && wri.isOpened() )
    while(1)
    {
        try{
            
            Mat frame;
            cap >> frame;

            // wri.write(frame);
            // printf("ok");
            // //imshow("lalala",frame);
            // int k = waitKey(10); 
            // if ( k==27 )
            //   break;
            
            //std::lock_guard<std::mutex> lock(mtx_mjpeg);
            static MJPG_sender wri(22222, 400000, 30);
            
            //cv::Mat mat = cv::cvarrToMat(ipl);
            //wri.write(*(cv::Mat*)frame);
            try{
                wri.write(frame);
            }catch (...) {
                std::cout << " wri.write error";
            }
            
            std::cout << " MJPEG-stream sent. \n";
            //utime.sleep(1)

        }catch (...) {
            VideoCapture cap;
            bool ok = cap.open("rtsp://admin:123456@192.168.10.11:554/h264/ch01/main/av_stream");

            std::cout << " Error in send_mjpeg() function \n";
        }
            
    }
    return 0;
}

打开浏览器，输入ip地址:22222，就可以看到推送的json流了