Chromium Code Reviews| OLD | NEW |
|---|---|
| 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. | 1 // Copyright (c) 2012 The Chromium Authors. All rights reserved. |
| 2 // Use of this source code is governed by a BSD-style license that can be | 2 // Use of this source code is governed by a BSD-style license that can be |
| 3 // found in the LICENSE file. | 3 // found in the LICENSE file. |
| 4 | 4 |
| 5 #include "media/base/callback_util.h" | 5 #include "media/base/callback_util.h" |
| 6 | 6 |
| 7 #include "base/bind.h" | 7 #include "base/bind.h" |
| 8 #include "base/synchronization/lock.h" | 8 #include "base/callback_helpers.h" |
| 9 #include "base/memory/ref_counted.h" | |
| 10 #include "base/message_loop.h" | 9 #include "base/message_loop.h" |
| 11 #include "base/message_loop_proxy.h" | 10 #include "base/message_loop_proxy.h" |
| 12 | 11 |
| 13 namespace media { | 12 namespace media { |
| 14 | 13 |
| 15 // Executes the given closure if and only if the closure returned by | 14 // Converts a bound function accepting a Closure into a bound function |
| 16 // GetClosure() has been executed exactly |count| times. | 15 // accepting a PipelineStatusCB. Since closures have no way of reporting a |
| 17 // | 16 // status |status_cb| is executed with PIPELINE_OK. |
| 18 // |done_cb| will be executed on the same thread that created the CountingCB. | 17 void RunBoundClosure( |
|
Ami GONE FROM CHROMIUM
2012/08/03 20:31:43
static
| |
| 19 class CountingCB : public base::RefCountedThreadSafe<CountingCB> { | 18 const SerialCallbackRunner::BoundClosure& bound_closure, |
| 20 public: | 19 const PipelineStatusCB& status_cb) { |
| 21 CountingCB(int count, const base::Closure& done_cb) | 20 bound_closure.Run(base::Bind(status_cb, PIPELINE_OK)); |
| 22 : message_loop_(base::MessageLoopProxy::current()), | 21 } |
| 23 count_(count), | |
| 24 done_cb_(done_cb) { | |
| 25 } | |
| 26 | 22 |
| 27 // Returns a closure bound to this object. | 23 // Runs |status_cb| with |last_status| on |message_loop|, trampolining if |
| 28 base::Closure GetClosure() { | 24 // necessary. |
| 29 return base::Bind(&CountingCB::OnCallback, this); | 25 static void RunOnMessageLoop( |
| 30 } | 26 const scoped_refptr<base::MessageLoopProxy>& message_loop, |
| 31 | 27 const PipelineStatusCB& status_cb, |
| 32 protected: | 28 PipelineStatus last_status) { |
| 33 friend class base::RefCountedThreadSafe<CountingCB>; | |
| 34 virtual ~CountingCB() {} | |
| 35 | |
| 36 private: | |
| 37 void OnCallback() { | |
| 38 { | |
| 39 base::AutoLock l(lock_); | |
| 40 count_--; | |
| 41 DCHECK_GE(count_, 0) << "CountingCB executed too many times"; | |
| 42 if (count_ != 0) | |
| 43 return; | |
| 44 } | |
| 45 | |
| 46 if (!message_loop_->BelongsToCurrentThread()) { | |
| 47 message_loop_->PostTask(FROM_HERE, done_cb_); | |
| 48 return; | |
| 49 } | |
| 50 | |
| 51 done_cb_.Run(); | |
| 52 } | |
| 53 | |
| 54 scoped_refptr<base::MessageLoopProxy> message_loop_; | |
| 55 base::Lock lock_; | |
| 56 int count_; | |
| 57 base::Closure done_cb_; | |
| 58 | |
| 59 DISALLOW_COPY_AND_ASSIGN(CountingCB); | |
| 60 }; | |
| 61 | |
| 62 static void OnSeriesCallback( | |
| 63 scoped_refptr<base::MessageLoopProxy> message_loop, | |
| 64 scoped_ptr<std::queue<ClosureFunc> > closures, | |
| 65 const base::Closure& done_cb) { | |
| 66 if (!message_loop->BelongsToCurrentThread()) { | 29 if (!message_loop->BelongsToCurrentThread()) { |
| 67 message_loop->PostTask(FROM_HERE, base::Bind( | 30 message_loop->PostTask(FROM_HERE, base::Bind( |
| 68 &OnSeriesCallback, message_loop, base::Passed(&closures), done_cb)); | 31 &RunOnMessageLoop, message_loop, status_cb, last_status)); |
| 32 return; | |
| 33 } | |
| 34 status_cb.Run(last_status); | |
| 35 } | |
| 36 | |
| 37 SerialCallbackRunner::Queue::Queue() {} | |
| 38 SerialCallbackRunner::Queue::~Queue() {} | |
| 39 | |
| 40 void SerialCallbackRunner::Queue::Push( | |
| 41 const BoundClosure& bound_closure) { | |
| 42 bound_fns_.push(base::Bind(&RunBoundClosure, bound_closure)); | |
| 43 } | |
| 44 | |
| 45 void SerialCallbackRunner::Queue::Push( | |
| 46 const BoundPipelineStatusCB& bound_status_cb) { | |
| 47 bound_fns_.push(bound_status_cb); | |
| 48 } | |
| 49 | |
| 50 SerialCallbackRunner::BoundPipelineStatusCB SerialCallbackRunner::Queue::Pop() { | |
| 51 BoundPipelineStatusCB bound_fn = bound_fns_.front(); | |
| 52 bound_fns_.pop(); | |
| 53 return bound_fn; | |
| 54 } | |
| 55 | |
| 56 SerialCallbackRunner::SerialCallbackRunner( | |
| 57 scoped_ptr<Queue> bound_fns, const PipelineStatusCB& done_cb) | |
| 58 : weak_this_(this), | |
| 59 message_loop_(base::MessageLoopProxy::current()), | |
| 60 bound_fns_(bound_fns.Pass()), | |
| 61 done_cb_(done_cb) { | |
| 62 message_loop_->PostTask(FROM_HERE, base::Bind( | |
| 63 &SerialCallbackRunner::RunNextInSeries, weak_this_.GetWeakPtr(), | |
| 64 PIPELINE_OK)); | |
| 65 } | |
| 66 | |
| 67 SerialCallbackRunner::~SerialCallbackRunner() {} | |
| 68 | |
| 69 scoped_ptr<SerialCallbackRunner> SerialCallbackRunner::Run( | |
| 70 scoped_ptr<Queue> bound_fns, | |
| 71 const PipelineStatusCB& done_cb) { | |
| 72 scoped_ptr<SerialCallbackRunner> callback_series( | |
| 73 new SerialCallbackRunner(bound_fns.Pass(), done_cb)); | |
| 74 return callback_series.Pass(); | |
| 75 } | |
| 76 | |
| 77 void SerialCallbackRunner::RunNextInSeries(PipelineStatus last_status) { | |
| 78 DCHECK(message_loop_->BelongsToCurrentThread()); | |
| 79 DCHECK(!done_cb_.is_null()); | |
| 80 | |
| 81 if (bound_fns_->empty() || last_status != PIPELINE_OK) { | |
| 82 base::ResetAndReturn(&done_cb_).Run(last_status); | |
| 69 return; | 83 return; |
| 70 } | 84 } |
| 71 | 85 |
| 72 if (closures->empty()) { | 86 BoundPipelineStatusCB bound_fn = bound_fns_->Pop(); |
| 73 done_cb.Run(); | 87 bound_fn.Run(base::Bind(&RunOnMessageLoop, message_loop_, base::Bind( |
| 74 return; | 88 &SerialCallbackRunner::RunNextInSeries, weak_this_.GetWeakPtr()))); |
| 75 } | |
| 76 | |
| 77 ClosureFunc cb = closures->front(); | |
| 78 closures->pop(); | |
| 79 cb.Run(base::Bind( | |
| 80 &OnSeriesCallback, message_loop, base::Passed(&closures), done_cb)); | |
| 81 } | |
| 82 | |
| 83 void RunInSeries(scoped_ptr<std::queue<ClosureFunc> > closures, | |
| 84 const base::Closure& done_cb) { | |
| 85 OnSeriesCallback(base::MessageLoopProxy::current(), | |
| 86 closures.Pass(), done_cb); | |
| 87 } | |
| 88 | |
| 89 static void OnStatusCallback( | |
| 90 scoped_refptr<base::MessageLoopProxy> message_loop, | |
| 91 scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs, | |
| 92 const PipelineStatusCB& done_cb, | |
| 93 PipelineStatus last_status) { | |
| 94 if (!message_loop->BelongsToCurrentThread()) { | |
| 95 message_loop->PostTask(FROM_HERE, base::Bind( | |
| 96 &OnStatusCallback, message_loop, base::Passed(&status_cbs), done_cb, | |
| 97 last_status)); | |
| 98 return; | |
| 99 } | |
| 100 | |
| 101 if (status_cbs->empty() || last_status != PIPELINE_OK) { | |
| 102 done_cb.Run(last_status); | |
| 103 return; | |
| 104 } | |
| 105 | |
| 106 PipelineStatusCBFunc status_cb = status_cbs->front(); | |
| 107 status_cbs->pop(); | |
| 108 status_cb.Run(base::Bind( | |
| 109 &OnStatusCallback, message_loop, base::Passed(&status_cbs), done_cb)); | |
| 110 } | |
| 111 | |
| 112 void RunInSeriesWithStatus( | |
| 113 scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs, | |
| 114 const PipelineStatusCB& done_cb) { | |
| 115 OnStatusCallback(base::MessageLoopProxy::current(), | |
| 116 status_cbs.Pass(), done_cb, PIPELINE_OK); | |
| 117 } | |
| 118 | |
| 119 void RunInParallel(scoped_ptr<std::queue<ClosureFunc> > closures, | |
| 120 const base::Closure& done_cb) { | |
| 121 if (closures->empty()) { | |
| 122 done_cb.Run(); | |
| 123 return; | |
| 124 } | |
| 125 | |
| 126 scoped_refptr<CountingCB> counting_cb = | |
| 127 new CountingCB(closures->size(), done_cb); | |
| 128 while (!closures->empty()) { | |
| 129 closures->front().Run(counting_cb->GetClosure()); | |
| 130 closures->pop(); | |
| 131 } | |
| 132 } | 89 } |
| 133 | 90 |
| 134 } // namespace media | 91 } // namespace media |
| OLD | NEW |