aboutsummaryrefslogtreecommitdiffhomepage
path: root/src/objective-c/RxLibrary/GRXConcurrentWriteable.m
blob: c26231349605821864dd5709fd8039978884692a (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
/*
 *
 * Copyright 2015 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

#import "GRXConcurrentWriteable.h"

#import <RxLibrary/GRXWriteable.h>

@interface GRXConcurrentWriteable ()
// This is atomic so that cancellation can nillify it from any thread.
@property(atomic, strong) id<GRXWriteable> writeable;
@end

@implementation GRXConcurrentWriteable {
  dispatch_queue_t _writeableQueue;
  // This ensures that writesFinishedWithError: is only sent once to the writeable.
  BOOL _alreadyFinished;
}

- (instancetype)init {
  return [self initWithWriteable:nil];
}

// Designated initializer
- (instancetype)initWithWriteable:(id<GRXWriteable>)writeable
                    dispatchQueue:(dispatch_queue_t)queue {
  if (self = [super init]) {
    _writeableQueue = queue;
    _writeable = writeable;
  }
  return self;
}

- (instancetype)initWithWriteable:(id<GRXWriteable>)writeable {
  return [self initWithWriteable:writeable
                   dispatchQueue:dispatch_get_main_queue()];
}

- (void)enqueueValue:(id)value completionHandler:(void (^)(void))handler {
  dispatch_async(_writeableQueue, ^{
    // We're racing a possible cancellation performed by another thread. To turn all already-
    // enqueued messages into noops, cancellation nillifies the writeable property. If we get it
    // before it's nil, we won the race.
    id<GRXWriteable> writeable = self.writeable;
    if (writeable) {
      [writeable writeValue:value];
      handler();
    }
  });
}

- (void)enqueueSuccessfulCompletion {
  __weak typeof(self) weakSelf = self;
  dispatch_async(_writeableQueue, ^{
    typeof(self) strongSelf = weakSelf;
    if (strongSelf) {
      BOOL finished = NO;
      @synchronized (self) {
        if (!strongSelf->_alreadyFinished) {
          strongSelf->_alreadyFinished = YES;
        } else {
          finished = YES;
        }
      }
      if (!finished) {
        // Cancellation is now impossible. None of the other three blocks can run concurrently with
        // this one.
        [self.writeable writesFinishedWithError:nil];
        // Skip any possible message to the wrapped writeable enqueued after this one.
        self.writeable = nil;
      }
    }
  });
}

- (void)cancelWithError:(NSError *)error {
  NSAssert(error, @"For a successful completion, use enqueueSuccessfulCompletion.");
  BOOL finished = NO;
  @synchronized (self) {
    if (!_alreadyFinished) {
      _alreadyFinished = YES;
    } else {
      finished = YES;
    }
  }
  if (!finished) {
    // Skip any of the still-enqueued messages to the wrapped writeable. We use the atomic setter to
    // nillify writeable because we might be running concurrently with the blocks in
    // _writeableQueue, and assignment with ARC isn't atomic.
    id<GRXWriteable> writeable = self.writeable;
    self.writeable = nil;

    dispatch_async(_writeableQueue, ^{
      [writeable writesFinishedWithError:error];
    });
  }
}

- (void)cancelSilently {
  BOOL finished = NO;
  @synchronized (self) {
    if (!_alreadyFinished) {
      _alreadyFinished = YES;
    } else {
      finished = YES;
    }
  }
  if (!finished) {
    // Skip any of the still-enqueued messages to the wrapped writeable. We use the atomic setter to
    // nillify writeable because we might be running concurrently with the blocks in
    // _writeableQueue, and assignment with ARC isn't atomic.
    self.writeable = nil;
  }
}
@end