Skip to content

Commit 7339548

Browse files
pasinpasin
committed
Merge pull request #650 from couchbase/feature/issue_645_test_failure_cbljava
Fixed #645 CBL Java Unit Test Failure on Jenkins
2 parents 61d0214 + 5cf092a commit 7339548

2 files changed

Lines changed: 77 additions & 73 deletions

File tree

‎src/main/java/com/couchbase/lite/LiveQuery.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -128,6 +128,9 @@ public void stop() {
128128

129129
/**
130130
* Blocks until the intial async query finishes. After this call either .rows or .error will be non-nil.
131+
*
132+
* TODO: It seems that implementation of waitForRows() is not correct. Should fix this!!!
133+
* https://github.com/couchbase/couchbase-lite-java-core/issues/647
131134
*/
132135
@InterfaceAudience.Public
133136
public void waitForRows() throws CouchbaseLiteException {
@@ -264,6 +267,8 @@ public void completed(QueryEnumerator rowsParam, Throwable error) {
264267
setRows(rowsParam);
265268
for (ChangeListener observer : observers) {
266269
Log.d(Log.TAG_QUERY, "%s: update() calling back observer with rows", LiveQuery.this);
270+
// TODO: LiveQuery.ChangeListener should not be fired for non-match?
271+
// https://github.com/couchbase/couchbase-lite-java-core/issues/648
267272
observer.changed(new ChangeEvent(LiveQuery.this, rows));
268273
}
269274
}

‎src/main/java/com/couchbase/lite/support/Batcher.java‎

Lines changed: 72 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -80,30 +80,32 @@ private boolean isCurrentlyProcessing() {
8080
*/
8181
public void queueObjects(List<T> objects) {
8282

83-
Log.v(Log.TAG_BATCHER, "%s: queueObjects called with %d objects. Thread: %s", this, objects.size(), Thread.currentThread());
84-
if (objects.size() == 0) {
85-
return;
86-
}
83+
synchronized (inbox) {
84+
Log.v(Log.TAG_BATCHER, "%s: queueObjects called with %d objects. Thread: %s", this, objects.size(), Thread.currentThread());
85+
if (objects.size() == 0) {
86+
return;
87+
}
8788

88-
Log.v(Log.TAG_BATCHER, "%s: inbox size before adding objects: %d", this, inbox.size());
89+
Log.v(Log.TAG_BATCHER, "%s: inbox size before adding objects: %d", this, inbox.size());
8990

90-
inbox.addAll(objects);
91+
inbox.addAll(objects);
9192

92-
if (inbox.size() >= capacity) {
93-
Log.v(Log.TAG_BATCHER, "%s: calling scheduleWithDelay(0)", this);
94-
if (!isCurrentlyProcessing()) {
95-
// we only want to unschedule and re-schedule immediately if we
96-
// aren't currently processing, otherwise we'll end up in a situation where
97-
// the things currently processed get interrupted and re-scheduled,
98-
// and nothing ever gets done. at the end of the processNow() method,
99-
// it will schedule tasks if more are still pending.
100-
unscheduleAllPending();
101-
scheduleWithDelay(0);
93+
if (inbox.size() >= capacity) {
94+
Log.v(Log.TAG_BATCHER, "%s: calling scheduleWithDelay(0)", this);
95+
if (!isCurrentlyProcessing()) {
96+
// we only want to unschedule and re-schedule immediately if we
97+
// aren't currently processing, otherwise we'll end up in a situation where
98+
// the things currently processed get interrupted and re-scheduled,
99+
// and nothing ever gets done. at the end of the processNow() method,
100+
// it will schedule tasks if more are still pending.
101+
unscheduleAllPending();
102+
scheduleWithDelay(0);
103+
}
104+
} else {
105+
int suggestedDelay = delayToUse();
106+
Log.v(Log.TAG_BATCHER, "%s: calling scheduleWithDelay(%d)", this, suggestedDelay);
107+
scheduleWithDelay(suggestedDelay);
102108
}
103-
} else {
104-
int suggestedDelay = delayToUse();
105-
Log.v(Log.TAG_BATCHER, "%s: calling scheduleWithDelay(%d)", this, suggestedDelay);
106-
scheduleWithDelay(suggestedDelay);
107109
}
108110
}
109111

@@ -146,10 +148,7 @@ public void flush() {
146148
}
147149

148150
public int count() {
149-
synchronized(this) {
150-
if(inbox == null) {
151-
return 0;
152-
}
151+
synchronized (inbox) {
153152
return inbox.size();
154153
}
155154
}
@@ -158,65 +157,65 @@ public int count() {
158157
* this is mainly for debugging
159158
*/
160159
public int sizeOfPendingFutures() {
161-
synchronized(this) {
162-
if(pendingFutures == null) {
163-
return 0;
164-
}
160+
synchronized (pendingFutures) {
165161
return pendingFutures.size();
166162
}
167163
}
168164

169165
private void processNow() {
170-
Log.v(Log.TAG_BATCHER, this + ": processNow() called");
171-
172-
List<T> toProcess = new ArrayList<T>();
173166

174-
if (inbox == null || inbox.size() == 0) {
175-
Log.v(Log.TAG_BATCHER, this + ": processNow() called, but inbox is empty");
176-
return;
177-
} else if (inbox.size() <= capacity) {
178-
Log.v(Log.TAG_BATCHER, "%s: inbox.size() <= capacity, adding %d items from inbox -> toProcess", this, inbox.size());
179-
while (inbox.size() > 0) {
180-
try {
181-
T t = inbox.take();
182-
toProcess.add(t);
183-
} catch (InterruptedException e) {
184-
Log.w(Log.TAG_BATCHER, "%s: processNow(): %s", this, e.getMessage());
167+
synchronized (inbox) {
168+
Log.v(Log.TAG_BATCHER, this + ": processNow() called");
169+
170+
List<T> toProcess = new ArrayList<T>();
171+
172+
if (inbox.size() == 0) {
173+
Log.v(Log.TAG_BATCHER, this + ": processNow() called, but inbox is empty");
174+
return;
175+
} else if (inbox.size() <= capacity) {
176+
Log.v(Log.TAG_BATCHER, "%s: inbox.size() <= capacity, adding %d items from inbox -> toProcess", this, inbox.size());
177+
while (inbox.size() > 0) {
178+
try {
179+
T t = inbox.take();
180+
toProcess.add(t);
181+
} catch (InterruptedException e) {
182+
Log.w(Log.TAG_BATCHER, "%s: processNow(): %s", this, e.getMessage());
183+
}
185184
}
186-
}
187-
} else {
188-
Log.v(Log.TAG_BATCHER, "%s: processNow() called, inbox size: %d", this, inbox.size());
189-
int i = 0;
190-
while (inbox.size() > 0 && i < capacity) {
191-
try {
192-
T t = inbox.take();
193-
toProcess.add(t);
194-
} catch (InterruptedException e) {
195-
Log.w(Log.TAG_BATCHER, "%s: processNow(): %s", this, e.getMessage());
185+
} else {
186+
Log.v(Log.TAG_BATCHER, "%s: processNow() called, inbox size: %d", this, inbox.size());
187+
int i = 0;
188+
while (inbox.size() > 0 && i < capacity) {
189+
try {
190+
T t = inbox.take();
191+
toProcess.add(t);
192+
} catch (InterruptedException e) {
193+
Log.w(Log.TAG_BATCHER, "%s: processNow(): %s", this, e.getMessage());
194+
}
195+
i += 1;
196196
}
197-
i += 1;
198-
}
199197

200-
Log.v(Log.TAG_BATCHER, "%s: inbox.size() > capacity, moving %d items from inbox -> toProcess array", this, toProcess.size());
201-
}
198+
Log.v(Log.TAG_BATCHER, "%s: inbox.size() > capacity, moving %d items from inbox -> toProcess array", this, toProcess.size());
199+
}
202200

203-
if (toProcess != null && toProcess.size() > 0) {
204-
Log.v(Log.TAG_BATCHER, "%s: invoking processor %s with %d items ", this, processor, toProcess.size());
205-
processor.process(toProcess);
206-
} else {
207-
Log.v(Log.TAG_BATCHER, "%s: nothing to process", this);
208-
}
209-
lastProcessedTime = System.currentTimeMillis();
210-
211-
// in case we ignored any schedule requests while processing, if
212-
// we have more items in inbox, lets schedule another processing attempt
213-
if (inbox.size() > 0) {
214-
Log.v(Log.TAG_BATCHER, "%s: finished processing a batch, but inbox size > 0: %d", this, inbox.size());
215-
//int delayToUse = delayToUse();
216-
int delayToUse = 0;
217-
Log.v(Log.TAG_BATCHER, "%s: going to process with delay: %d", this, delayToUse);
218-
ScheduledFuture pendingFuture = workExecutor.schedule(processNowRunnable, delayToUse, TimeUnit.MILLISECONDS);
219-
pendingFutures.add(pendingFuture);
201+
if (toProcess != null && toProcess.size() > 0) {
202+
Log.v(Log.TAG_BATCHER, "%s: invoking processor %s with %d items ", this, processor, toProcess.size());
203+
processor.process(toProcess);
204+
} else {
205+
Log.v(Log.TAG_BATCHER, "%s: nothing to process", this);
206+
}
207+
lastProcessedTime = System.currentTimeMillis();
208+
209+
// in case we ignored any schedule requests while processing, if
210+
// we have more items in inbox, lets schedule another processing attempt
211+
if (inbox.size() > 0) {
212+
Log.v(Log.TAG_BATCHER, "%s: finished processing a batch, but inbox size > 0: %d", this, inbox.size());
213+
//int delayToUse = delayToUse();
214+
int delayToUse = 0;
215+
Log.v(Log.TAG_BATCHER, "%s: going to process with delay: %d", this, delayToUse);
216+
ScheduledFuture pendingFuture = workExecutor.schedule(processNowRunnable, delayToUse, TimeUnit.MILLISECONDS);
217+
pendingFutures.add(pendingFuture);
218+
}
220219
}
221220
}
222221

0 commit comments

Comments
 (0)