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 void RunClosureFunc(const ClosureFunc& closure, |
16 // GetClosure() has been executed exactly |count| times. | 15 const PipelineStatusCB& status_cb) { |
17 // | 16 closure.Run(base::Bind(status_cb, PIPELINE_OK)); |
18 // |done_cb| will be executed on the same thread that created the CountingCB. | 17 } |
19 class CountingCB : public base::RefCountedThreadSafe<CountingCB> { | |
20 public: | |
21 CountingCB(int count, const base::Closure& done_cb) | |
22 : message_loop_(base::MessageLoopProxy::current()), | |
23 count_(count), | |
24 done_cb_(done_cb) { | |
25 } | |
26 | 18 |
27 // Returns a closure bound to this object. | 19 // Runs |status_cb| with |last_status| on |message_loop|, trampolining if |
28 base::Closure GetClosure() { | 20 // necessary. |
29 return base::Bind(&CountingCB::OnCallback, this); | 21 static void RunOnMessageLoop( |
30 } | 22 const scoped_refptr<base::MessageLoopProxy>& message_loop, |
31 | 23 const PipelineStatusCB& status_cb, |
32 protected: | 24 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()) { | 25 if (!message_loop->BelongsToCurrentThread()) { |
67 message_loop->PostTask(FROM_HERE, base::Bind( | 26 message_loop->PostTask(FROM_HERE, base::Bind( |
68 &OnSeriesCallback, message_loop, base::Passed(&closures), done_cb)); | 27 &RunOnMessageLoop, message_loop, status_cb, last_status)); |
| 28 return; |
| 29 } |
| 30 status_cb.Run(last_status); |
| 31 } |
| 32 |
| 33 CallbackSeries::CallbackSeries( |
| 34 scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs, |
| 35 const PipelineStatusCB& done_cb) |
| 36 : weak_this_(this), |
| 37 message_loop_(base::MessageLoopProxy::current()), |
| 38 status_cbs_(status_cbs.Pass()), |
| 39 done_cb_(done_cb) { |
| 40 } |
| 41 |
| 42 CallbackSeries::~CallbackSeries() {} |
| 43 |
| 44 scoped_ptr<CallbackSeries> CallbackSeries::Run( |
| 45 scoped_ptr<std::queue<PipelineStatusCBFunc> > status_cbs, |
| 46 const PipelineStatusCB& done_cb) { |
| 47 scoped_ptr<CallbackSeries> callback_series( |
| 48 new CallbackSeries(status_cbs.Pass(), done_cb)); |
| 49 callback_series->RunNextInSeries(PIPELINE_OK); |
| 50 return callback_series.Pass(); |
| 51 } |
| 52 |
| 53 void CallbackSeries::RunNextInSeries(PipelineStatus last_status) { |
| 54 DCHECK(message_loop_->BelongsToCurrentThread()); |
| 55 DCHECK(!done_cb_.is_null()); |
| 56 |
| 57 if (status_cbs_->empty() || last_status != PIPELINE_OK) { |
| 58 base::ResetAndReturn(&done_cb_).Run(last_status); |
69 return; | 59 return; |
70 } | 60 } |
71 | 61 |
72 if (closures->empty()) { | 62 PipelineStatusCBFunc status_cb = status_cbs_->front(); |
73 done_cb.Run(); | 63 status_cbs_->pop(); |
74 return; | 64 status_cb.Run(base::Bind(&RunOnMessageLoop, message_loop_, base::Bind( |
75 } | 65 &CallbackSeries::RunNextInSeries, weak_this_.GetWeakPtr()))); |
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 } | 66 } |
133 | 67 |
134 } // namespace media | 68 } // namespace media |
OLD | NEW |