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 |