ffplay: add msg_queue

ijkmediaplayer/jni/ffplay/ff_ffmsg_queue.h

+/*
+ * ff_ffmsg_queue.h
+ *      based on PacketQueue in ffplay.c
+ *
+ * Copyright (c) 2013 Zhang Rui <bbcallen@gmail.com>
+ *
+ * This file is part of ijkPlayer.
+ *
+ * ijkPlayer is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License as published by the Free Software Foundation; either
+ * version 2.1 of the License, or (at your option) any later version.
+ *
+ * ijkPlayer is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with ijkPlayer; if not, write to the Free Software
+ * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
+ */
+
+#ifndef FFPLAY__FF_FFMSG_QUEUE_H
+#define FFPLAY__FF_FFMSG_QUEUE_H
+
+#include "ff_ffinc.h"
+#include "ff_ffmsg.h"
+
+typedef struct AVMessage {
+    int what;
+    int arg1;
+    int arg2;
+    struct AVMessage *next;
+} AVMessage;
+
+typedef struct MessageQueue {
+    AVMessage *first_msg, *last_msg;
+    int nb_messages;
+    int abort_request;
+    SDL_mutex *mutex;
+    SDL_cond *cond;
+} MessageQueue;
+
+// TODO: msg pool
+inline static int msg_queue_put_private(MessageQueue *q, AVMessage *msg)
+{
+    AVMessage *msg1;
+
+    if (q->abort_request)
+        return -1;
+
+    msg1 = av_malloc(sizeof(AVMessage));
+    if (!msg1)
+        return -1;
+    *msg1 = *msg;
+    msg1->next = NULL;
+
+    if (!q->last_msg)
+        q->first_msg = msg1;
+    else
+        q->last_msg->next = msg1;
+    q->last_msg = msg1;
+    q->nb_messages++;
+    SDL_CondSignal(q->cond);
+    return 0;
+}
+
+inline static int msg_queue_put(MessageQueue *q, AVMessage *msg)
+{
+    int ret;
+
+    SDL_LockMutex(q->mutex);
+    ret = packet_queue_put_private(q, msg);
+    SDL_UnlockMutex(q->mutex);
+
+    return ret;
+}
+
+inline static void msg_queue_init(MessageQueue *q)
+{
+    memset(q, 0, sizeof(MessageQueue));
+    q->mutex = SDL_CreateMutex();
+    q->cond = SDL_CreateCond();
+    q->abort_request = 1;
+}
+
+inline static void msg_queue_flush(MessageQueue *q)
+{
+    AVMessage *msg, *msg1;
+
+    SDL_LockMutex(q->mutex);
+    for (msg = q->first_msg; msg != NULL; msg = msg1) {
+        msg1 = msg->next;
+        av_freep(&msg);
+    }
+    q->last_msg = NULL;
+    q->first_msg = NULL;
+    q->nb_messages = 0;
+    SDL_UnlockMutex(q->mutex);
+}
+
+inline static void msg_queue_destroy(MessageQueue *q)
+{
+    msg_queue_flush(q);
+    SDL_DestroyMutex(q->mutex);
+    SDL_DestroyCond(q->cond);
+}
+
+inline static void msg_queue_abort(MessageQueue *q)
+{
+    SDL_LockMutex(q->mutex);
+
+    q->abort_request = 1;
+
+    SDL_CondSignal(q->cond);
+
+    SDL_UnlockMutex(q->mutex);
+}
+
+inline static void msg_queue_start(MessageQueue *q)
+{
+    SDL_LockMutex(q->mutex);
+    q->abort_request = 0;
+    msg_queue_put_simple1(q, FFP_MSG_FLUSH);
+    SDL_UnlockMutex(q->mutex);
+}
+
+/* return < 0 if aborted, 0 if no packet and > 0 if packet.  */
+static int msg_queue_get(MessageQueue *q, AVMessage *msg, int block)
+{
+    AVMessage *msg1;
+    int ret;
+
+    SDL_LockMutex(q->mutex);
+
+    for (;;) {
+        if (q->abort_request) {
+            ret = -1;
+            break;
+        }
+
+        msg1 = q->first_msg;
+        if (msg1) {
+            q->first_msg = msg1->next;
+            if (!q->first_msg)
+                q->last_msg = NULL;
+            q->nb_messages--;
+            *msg = *msg1;
+            av_free(msg1);
+            ret = 1;
+            break;
+        } else if (!block) {
+            ret = 0;
+            break;
+        } else {
+            SDL_CondWait(q->cond, q->mutex);
+        }
+    }
+    SDL_UnlockMutex(q->mutex);
+    return ret;
+}
+
+inline static void msg_init_msg(AVMessage *msg)
+{
+    memset(msg, 0, sizeof(AVMessage));
+}
+
+inline static void msg_queue_put_simple1(MessageQueue *q, int what)
+{
+    AVMessage msg;
+    msg_init_msg(&msg);
+    msg.what = what;
+    msg_queue_put(msg);
+}
+
+inline static void msg_queue_put_simple2(MessageQueue *q, int what, int arg1)
+{
+    AVMessage msg;
+    msg_init_msg(&msg);
+    msg.what = what;
+    msg.arg1 = arg1;
+    msg_queue_put(q, &msg);
+}
+
+inline static void msg_queue_put_simple3(MessageQueue *q, int what, int arg1, int arg2)
+{
+    AVMessage msg;
+    msg_init_msg(&msg);
+    msg.what = what;
+    msg.arg1 = arg1;
+    msg.arg2 = arg2;
+    msg_queue_put(q, &msg);
+}
+
+#endif

ijkmediaplayer/jni/ffplay/ff_ffplay.c

@@ -1689,7 +1689,7 @@ static int read_thread(void *arg)
         ffp->infinite_buffer = 1;
 
     prepared = true;
-    ffp_notify_msg(ffp, FFP_MSG_PREPARED, 0, 0, NULL);
+    ffp_notify_msg(ffp, FFP_MSG_PREPARED, 0, 0);
 
     for (;;) {
         if (is->abort_request)
@@ -1800,7 +1800,7 @@ static int read_thread(void *arg)
                     ret = AVERROR_EOF;
                     goto fail;
                 } else {
-                    ffp_notify_msg(ffp, FFP_MSG_COMPLETED, 0, 0, NULL);
+                    ffp_notify_msg(ffp, FFP_MSG_COMPLETED, 0, 0);
                 }
             }
             eof=0;
@@ -1859,7 +1859,7 @@ fail:
 
     if (!prepared || !is->abort_request) {
         ffp->last_error = last_error;
-        ffp_notify_msg(ffp, FFP_MSG_ERROR, last_error, 0, NULL);
+        ffp_notify_msg(ffp, FFP_MSG_ERROR, last_error, 0);
     }
 
     SDL_DestroyMutex(wait_mutex);
@@ -2053,6 +2053,8 @@ FFPlayer *ffp_create_ffplayer()
     if (!ffp)
         return NULL;
 
+    msg_queue_init(&ffp->msg_queue);
+
     memset(ffp, 0, sizeof(FFPlayer));
     ffp_reset_internal(ffp);
     return ffp;
@@ -2071,6 +2073,8 @@ void ffp_destroy_ffplayer(FFPlayer **pffp)
     }
 
     ffp_reset_internal(ffp);
+
+    msg_queue_destroy(&ffp->msg_queue);
     free(ffp);
     *pffp = NULL;
 }
@@ -2134,6 +2138,7 @@ int ffp_stop_l(FFPlayer *ffp)
     if (!is)
         return EIJK_NULL_IS_PTR;
 
+    msg_queue_abort(&ffp->msg_queue);
     is->abort_request = 1;
     return 0;
 }
@@ -2141,10 +2146,8 @@ int ffp_stop_l(FFPlayer *ffp)
 int ffp_wait_stop_l(FFPlayer *ffp)
 {
     assert(ffp);
-    VideoState *is = ffp->is;
-    if (!is)
-        return EIJK_NULL_IS_PTR;
 
+    ffp_stop_l(ffp);
     stream_close(ffp);
     ffp->is = NULL;
     return 0;

ijkmediaplayer/jni/ffplay/ff_ffplay_def.h

@@ -26,6 +26,7 @@
 
 #include "ff_ffinc.h"
 #include "ff_ffplay_config.h"
+#include "ff_ffmsg_queue.h"
 
 #define MAX_QUEUE_SIZE (15 * 1024 * 1024)
 #define MIN_FRAMES 5
@@ -397,8 +398,7 @@ typedef struct FFPlayer {
 
     int last_error;
 
-    void  *msg_opaque;
-    void (*msg_handler)(void *opaque, int what, int arg1, int arg2, void *data);
+    MessageQueue msg_queue;
 } FFPlayer;
 
 #define FFP_SAFE_FREE(p) do {free(p); p = NULL;} while(0)
@@ -462,13 +462,11 @@ inline static void ffp_reset_internal(FFPlayer *ffp)
 
     ffp->last_error             = 0;
 
-    ffp->msg_opaque             = 0;
-    ffp->msg_handler            = NULL;
+    msg_queue_flush(&ffp->msg_queue);
 }
 
-inline static void ffp_notify_msg(FFPlayer *ffp, int what, int arg1, int arg2, void* data) {
-    if (ffp->msg_handler)
-        ffp->msg_handler(ffp->msg_opaque, what, arg1, arg2, data);
+inline static void ffp_notify_msg(FFPlayer *ffp, int what, int arg1, int arg2) {
+    msg_queue_put_simple3(&ffp->msg_queue, what, arg1, arg2);
 }
 
 #endif

ijkmediaplayer/jni/ijkplayer/android/ijkplayer_android.c

@@ -39,8 +39,6 @@ typedef struct IjkMediaPlayer {
     pthread_mutex_t mutex;
     FFPlayer *ffplayer;
 
-    // FIXME: IjkMessageQueue msg_queue;
-
     int mp_state;
     char *data_source;
 } IjkMediaPlayer;
@@ -61,7 +59,6 @@ inline static void destroy_mp(IjkMediaPlayer **pmp)
         ffp_destroy_ffplayer(&mp->ffplayer);
     }
 
-    // FIXME: ijkmsg_queue_destroy(&mp->msg_queue);
     pthread_mutex_destroy(&mp->mutex);
 
     FFP_SAFE_FREE(mp->data_source);
@@ -80,23 +77,6 @@ void ijkmp_global_uninit()
     ffp_global_uninit();
 }
 
-static void ijkmp_msg_handler(void *opaque, int what, int arg1, int arg2, void* data)
-{
-    // IjkMediaPlayer *mp = (IjkMediaPlayer *) opaque;
-    // FFPlayer *ffp = mp->ffplayer;
-
-    // FIXME: implement
-}
-
-void ijkmp_setup_internal(IjkMediaPlayer *mp) {
-    FFPlayer *ffp = mp->ffplayer;
-
-    ffp->msg_opaque = mp;
-    ffp->msg_handler = ijkmp_msg_handler;
-    // FIXME: ijkmsg_queue_flush(&mp->msg_queue);
-    // FIXME: ijkmsg_queue_start(&mp->msg_queue);
-}
-
 IjkMediaPlayer *ijkmp_create()
 {
     IjkMediaPlayer *mp = (IjkMediaPlayer *) malloc(sizeof(IjkMediaPlayer));
@@ -112,12 +92,10 @@ IjkMediaPlayer *ijkmp_create()
     }
 
     pthread_mutex_init(&mp->mutex, NULL);
-    // FIXME: ijkmsg_queue_init(&mp->msg_queue);
 
     ijkmp_inc_ref(mp);
 
     mp->ffplayer = ffp;
-    ijkmp_setup_internal(mp);
 
     return mp;
 }
@@ -126,8 +104,6 @@ void ijkmp_shutdown_l(IjkMediaPlayer *mp)
 {
     assert(mp);
 
-    // FIXME: ijkmsg_queue_abort(&mp->msg_queue);
-
     if (mp->ffplayer) {
         ffp_stop_l(mp->ffplayer);
         ffp_wait_stop_l(mp->ffplayer);
@@ -148,8 +124,6 @@ void ijkmp_reset_l(IjkMediaPlayer *mp)
 
     FFP_SAFE_FREE(mp->data_source);
     mp->mp_state = MP_STATE_IDLE;
-
-    ijkmp_setup_internal(mp);
 }
 
 void ijkmp_reset(IjkMediaPlayer *mp)
@@ -234,6 +208,7 @@ static int ijkmp_prepare_async_l(IjkMediaPlayer *mp)
     assert(mp->data_source);
 
     mp->mp_state = MP_STATE_ASYNC_PREPARING;
+    msg_queue_start(&mp->ffplayer->msg_queue);
     int retval = ffp_prepare_async_l(mp->ffplayer, mp->data_source);
     if (retval < 0) {
         mp->mp_state = MP_STATE_ERROR;