使用带有 ffmpeg 管道的 Windows 命名管道

Question

我正在尝试使用 Windows 管道将数据写入 FFmpeg 中的输入管道。我正在为 FFmpeg 使用以下命令：

ffmpeg -r 24 -pix_fmt rgba -s 1280x720 -f rawvideo -y -i \\.\pipe\videopipe -f s16le -ac 1 -ar 44100 -i \\.\pipe\audiopipe -acodec pcm_s16le -ac 1 -b:a 320k -ar 44100 -vf vflip -vcodec mpeg1video -qscale 4 -bufsize 500KB -maxrate 5000KB OUTPUT_FILE

我尝试使用CreateFile() 方法连接到它，但这似乎不起作用。在开始运行 ffmpeg 命令后，我也尝试过使用CreateNamedPipe()，但它似乎在等待其他连接。

我不知道我必须按什么顺序调用这两个。我是否首先需要创建一个 Windows 管道并在 FFmpeg 中使用相同的名称，还是需要先使用命名管道调用 FFmpeg，然后使用CreateFile() 连接到它？

Answer 1

我已经开发了代码来做到这一点，也许它会有所帮助。就我而言，我正在流式传输图像并在 ffplay 中播放。我启动了 3 个线程，一个创建管道（在演示管道中执行“initializePipe”的那个），一个监听管道创建并启动 ffplay 以及将图像写入管道的写入器线程。

demoPipes.cpp

#include "demoPipes.h"
using namespace std;

#define MAX_THREADS 100
//#define BUF_SIZE 255

void ErrorHandler(LPTSTR lpszFunction);

// Sample custom data structure for threads to use.
// This is passed by void pointer so it can be any data type
// that can be passed using a single void pointer (LPVOID).
/*typedef struct MyData {
    int val1;
    int val2;
} MYDATA, *PMYDATA;
*/

typedef struct
{
    HANDLE pipe;
    LPCSTR MEDIA_PIPE;
} Session;


int _tmain()
{
    Session* s = (Session*)malloc(sizeof(Session));

    s->pipe = NULL;
    s->MEDIA_PIPE = "\\\\.\\pipe\\screenRec";

    DWORD   dwThreadIdArray[MAX_THREADS];
    HANDLE  hThreadArray[MAX_THREADS]; 

    // Create MAX_THREADS worker threads.
    for( int i=0; i<MAX_THREADS; i++ )
    {
        //create pipe initializer thread
        if(i==0){
            hThreadArray[i] = CreateThread( 
                NULL,                   // default security attributes
                0,                      // use default stack size  
                initializePipe,       // thread function name
                (LPVOID) s,          // argument to thread function 
                0,                      // use default creation flags 
                &dwThreadIdArray[i]);   // returns the thread identifier 
        }

        //create writer thread
        if(i>1){
            PipeMetaData* pipe_data1 = fetch_file("D:\\images.png", s->pipe);
            //PipeMetaData* pipe_data = createPipeMetaData(pipe_data1->file_vector,pipe_data1->byte_size);
            //pipe_data1->pipe =s->pipe; 
            hThreadArray[i] = CreateThread( 
                NULL,                   // default security attributes
                0,                      // use default stack size  
                send_media_to_recorder,       // thread function name
                pipe_data1,          // argument to thread function 
                0,                      // use default creation flags 
                &dwThreadIdArray[i]);   // returns the thread identifier 
        }
        //create recorder thread
        if(i==1){
            hThreadArray[i] = CreateThread( 
                NULL,                   // default security attributes
                0,                      // use default stack size  
                recorder_pipe_function,       // thread function name
                0,          // argument to thread function 
                0,                      // use default creation flags 
                &dwThreadIdArray[i]);   // returns the thread identifier 
        }

        // Check the return value for success.
        // If CreateThread fails, terminate execution. 
        // This will automatically clean up threads and memory. 

        if (dwThreadIdArray[i] == NULL) 
        {
           ErrorHandler(TEXT("CreateThread"));
           ExitProcess(3);
        }
    } // End of main thread creation loop.


    //for close RTP
    // Wait until all threads have terminated.
    WaitForMultipleObjects(MAX_THREADS, hThreadArray, TRUE, INFINITE);

    // Close all thread handles and free memory allocations.
    CloseHandle(s->pipe);
    for(int i=0; i<MAX_THREADS; i++)
    {
        CloseHandle(hThreadArray[i]);
    }
    getchar();
    return 0;
}


void ErrorHandler(LPTSTR lpszFunction) 
{ 
    // Retrieve the system error message for the last-error code.

    LPVOID lpMsgBuf;
    LPVOID lpDisplayBuf;
    DWORD dw = GetLastError(); 

    FormatMessage(
        FORMAT_MESSAGE_ALLOCATE_BUFFER | 
        FORMAT_MESSAGE_FROM_SYSTEM |
        FORMAT_MESSAGE_IGNORE_INSERTS,
        NULL,
        dw,
        MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
        (LPTSTR) &lpMsgBuf,
        0, NULL );

    // Display the error message.

    lpDisplayBuf = (LPVOID)LocalAlloc(LMEM_ZEROINIT, 
        (lstrlen((LPCTSTR) lpMsgBuf) + lstrlen((LPCTSTR) lpszFunction) + 40) * sizeof(TCHAR)); 
    StringCchPrintf((LPTSTR)lpDisplayBuf, 
        LocalSize(lpDisplayBuf) / sizeof(TCHAR),
        TEXT("%s failed with error %d: %s"), 
        lpszFunction, dw, lpMsgBuf); 
    MessageBox(NULL, (LPCTSTR) lpDisplayBuf, TEXT("Error"), MB_OK); 

    // Free error-handling buffer allocations.

    LocalFree(lpMsgBuf);
    LocalFree(lpDisplayBuf);
}

clientpipe.cpp

#include "client_pipe.h"
using namespace std;

DWORD WINAPI recorder_pipe_function( LPVOID lpParam){
    const char* MEDIA_PIPE =  "\\\\.\\pipe\\screenRec";
    //LPCWSTR MEDIA_PIPE_wchar = L"\\\\.\\pipe\\screenRec";

    //waits for server to initialize the media pipe.
    //wait_for_media_pipe(MEDIA_PIPE,10);
    WaitNamedPipeA(MEDIA_PIPE,1000);
    start_recording_session(MEDIA_PIPE);

    return 0;
}

void start_recording_session(const char* media_pipe){
    //initialize recording command
    char* command = (char*)malloc(sizeof(char)*100+1);
    command[0] ='\0';
    strcpy_s(command,sizeof(char)*100+1,"start ffplay -i ");
    //strcpy_s(command,sizeof(char)*100+1,"ffmpeg -i ");
    strcat_s(command,sizeof(char)*100+1, media_pipe);
    //strcat_s(command,sizeof(char)*100+1, " -f matroska D:\\djhfifj.mkv");  
    std::cout << "Executing ffmpeg command\n";
    int i=system (command);
    std::cout << "The value returned was: " << i << std::endl;
}

ServerPipe.cpp

#include "server_pipe.h"

using namespace std;

typedef struct
{
    HANDLE pipe;
    LPCSTR MEDIA_PIPE;
} Session;

DWORD WINAPI initializePipe( LPVOID lpParam ) {

    printf("Creating an instance of a named pipe...");

    Session* s = (Session*) lpParam;
    LPCSTR MEDIA_PIPE = s->MEDIA_PIPE;

    // Create a pipe to send data
    s->pipe = CreateNamedPipeA(
        MEDIA_PIPE, // name of the pipe
        PIPE_ACCESS_OUTBOUND, // 1-way pipe -- send only
        PIPE_TYPE_BYTE, // send data as a byte stream
        1, // only allow 1 instance of this pipe
        0, // no outbound buffer
        0, // no inbound buffer
        0, // use default wait time
        NULL // use default security attributes
    );

    if (s->pipe == NULL ||s->pipe == INVALID_HANDLE_VALUE) {
        wcout << "Failed to create outbound pipe instance.";
        // look up error code here using GetLastError()
        return NULL;
    }

    wcout << "Waiting for a client to connect to the pipe..." << endl;

    // This call blocks until a client process connects to the pipe
    BOOL result = ConnectNamedPipe(s->pipe, NULL);
    if (!result) {
        wcout << "Failed to make connection on named pipe." << endl;
        // look up error code here using GetLastError()
        CloseHandle(s->pipe); // close the pipe
        system("pause");
        return NULL;
    }
}

PipeMetaData* fetch_file(char* file_path,HANDLE pipe){
    PipeMetaData* data = (PipeMetaData*) malloc(sizeof(PipeMetaData));

    //opening file
    ifstream infile;
    infile.open(file_path,std::ios::binary);
    infile.seekg(0,std::ios::end);

    //get file byte size
    data->byte_size = infile.tellg();

    //fetch data to send
    data->file_vector =(char*) malloc(sizeof(char)*data->byte_size);
    infile.seekg(0,std::ios::beg);
    infile.read(&(data->file_vector)[0],data->byte_size);
    data->packet = &(data->file_vector)[0];
    data->pipe = pipe;
    wcout<<data->byte_size<<endl;

    return data;
}

DWORD WINAPI send_media_to_recorder(LPVOID lpParam){
    PipeMetaData* p = (PipeMetaData*) lpParam;
    // This call blocks until a client process reads all the data
    DWORD numBytesWritten = 0;
    BOOL result = WriteFile(
        p->pipe, // handle to our outbound pipe
        p->packet,//&(p->file_vector)[0], // data to send
        p->byte_size, // length of data to send (bytes)
        &numBytesWritten, // will store actual amount of data sent
        NULL // not using overlapped IO
    );

    if (result) {
        wcout << "Number of bytes sent: " << numBytesWritten << endl;
    } else {
        wcout << "Failed to send data." << endl;
        // look up error code here using GetLastError()
        GetLastError();
    }

    free(p);
    return 0;
}