RxJava中的常用操作符

创建操作

• create
• defer
  直到有观察者订阅时才创建Observable，并且为每个观察者创建一个新的Observable。
  defer.c.png
  Defer操作符会一直等待直到有观察者订阅它，然后它使用Observable工厂方法生成一个Observable。它对每个观察者都这样做，因此尽管每个订阅者都以为自己订阅的是同一个Observable，事实上每个订阅者获取的是它们自己的单独的数据序列。
  在某些情况下，等待直到最后一分钟（就是直到订阅发生时）才生成Observable可以确保Observable包含最新的数据。
• Empty/Never/Throw
  Empty
  创建一个不发射任何数据但是正常终止的Observable
  Never
  创建一个不发射数据也不终止的Observable
  Throw
  创建一个不发射数据以一个错误终止的Observable
• fromArray/fromIterable
  将其它种类的对象和数据类型转换为Observable

from.png

在RxJava中，from操作符可以转换Iterable和Array等。对于Iterable和数组，产生的Observable会发射Iterable或数组

• Interval
  创建一个按固定时间间隔发射无线递增的整数序列的Observable,RxJava将这个操作符实现为interval方法。它接受一个表示时间间隔的参数和一个表示时间单位的参数。

interval(long,TimeUnit)
interval(long,TimeUnit,Scheduler)

• just
  创建一个发射指定值的Observable,Just将单个数据转换为发射那个数据的Observable。
  Just类似于From，但是From会将数组或Iterable的数据取出然后逐个发射，而Just只是简单的原样发射，将数组或Iterable当做单个数据。just方法最多接受10个参数,返回一个按参数列表顺序发射这些数据的Observable
• Range
  创建一个发射特定整数序列的Observable
  RxJava将这个操作符实现为range函数，它接受两个参数，一个是范围的起始值，一个是范围的数据的数目。如果你将第二个参数设为0，将导致Observable不发射任何数据（如果设置为负数，会抛异常）。
• Repeat
  创建一个发射特定数据重复多次的Observable
  RxJava将这个操作符实现为repeat方法。它不是创建一个Observable，而是重复发射原始Observable的数据序列，这个序列或者是无限的，或者通过repeat(n)指定重复次数

变换操作

• map
  对Observable发射的每一项数据应用一个函数，执行变换操作

map.png

Map操作符对原始Observable发射的每一项数据应用一个你选择的函数，然后返回一个发射这些结果的Observable。
RxJava将这个操作符实现为map函数。这个操作符默认不在任何特定的调度器上执行。

Observable.just(1,2,3)
                .map(new Function<Integer, String>() {
                    @Override
                    public String apply(@NonNull Integer integer) throws Exception {
                        return "integer is" + integer;
                    }
                })
                .subscribe(new Consumer<String>() {
                    @Override
                    public void accept(String s) throws Exception {
                        Log.i(TAG, "accept: s = " + s);     
                    }
                });

08-24 17:08:19.573 27421-27421/com.example.wty.learnrxjava I/MainActivity: accept: s = integer is 1
08-24 17:08:19.573 27421-27421/com.example.wty.learnrxjava I/MainActivity: accept: s = integer is 2
08-24 17:08:19.573 27421-27421/com.example.wty.learnrxjava I/MainActivity: accept: s = integer is 3

• flatMap
  FlatMap将一个发射数据的Observable变换为多个Observables，然后将它们发射的数据合并后放进一个单独的Observable

mergeMap.png

FlatMap操作符使用一个指定的函数对原始Observable发射的每一项数据执行变换操作，这个函数返回一个本身也发射数据的Observable，然后FlatMap合并这些Observables发射的数据，最后将合并后的结果当做它自己的数据序列发射。
这个方法是很有用的，例如，当你有一个这样的Observable：它发射一个数据序列，这些数据本身包含Observable成员或者可以变换为Observable，因此你可以创建一个新的Observable发射这些次级Observable发射的数据的完整集合。
注意：FlatMap对这些Observables发射的数据做的是合并(merge)操作，因此它们可能是交错的。如果想要按照严格的顺序发射这些数据,使用ConcatMap操作符即可
举个例子:

Observable.fromIterable(getData())
                .flatMap(new Function<NoteBook, ObservableSource<Note>>() {
                    @Override
                    public ObservableSource<Note> apply(@NonNull NoteBook noteBook) throws Exception {
                        return Observable.fromIterable(noteBook.getNotes());
                    }
                })
                .subscribe(new Consumer<Note>() {
                    @Override
                    public void accept(Note note) throws Exception {
                        Log.i(TAG, "accept: " + note);
                    }
                });

08-24 17:14:09.512 32091-32091/? I/MainActivity: accept: Note{id='1', noteBookId='1', title='Introduction', content='$$$$$$$$$$$$$$$$$$'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='2', noteBookId='1', title='ReactiveX', content='########################'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='3', noteBookId='1', title='Observables', content='@@@@@@@@@@@@@@@@@@@@@@@@@@'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='4', noteBookId='1', title='Operators Categories', content='********************'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='5', noteBookId='1', title='RxJava文档和教程', content='********************'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='6', noteBookId='2', title='Retrofit入门教程1', content='$$$$$$$$$$$$$$$$$$'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='7', noteBookId='2', title='Retrofit入门教程2', content='########################'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='8', noteBookId='2', title='Retrofit入门教程3', content='@@@@@@@@@@@@@@@@@@@@@@@@@@'}
08-24 17:14:09.513 32091-32091/? I/MainActivity: accept: Note{id='9', noteBookId='2', title='Retrofit入门教程4', content='********************'}
08-24 17:14:09.514 32091-32091/? I/MainActivity: accept: Note{id='10', noteBookId='2', title='Retrofit入门教程5', content='********************'}

• Buffer
  定期收集Observable的数据放进一个数据包裹，然后发射这些数据包裹，而不是一次发射一个值。
  buffer.png
  Buffer操作符将一个Observable变换为另一个，原来的Observable正常发射数据，变换产生的Observable发射这些数据的缓存集合。Buffer操作符在很多语言特定的实现中有很多种变体，它们在如何缓存这个问题上存在区别。
  注意：如果原来的Observable发射了一个onError通知，Buffer会立即传递这个通知，而不是首先发射缓存的数据，即使在这之前缓存中包含了原始Observable发射的数据。

Observable.interval(1000,TimeUnit.MILLISECONDS)
                .buffer(5)
                .subscribe(new Consumer<List<Long>>() {
                    @Override
                    public void accept(List<Long> longs) throws Exception {
                        Log.i(TAG, "accept: longs = " + longs);
                    }
                });

08-24 17:04:12.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [0, 1, 2, 3, 4]
08-24 17:04:17.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [5, 6, 7, 8, 9]
08-24 17:04:22.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [10, 11, 12, 13, 14]
08-24 17:04:27.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [15, 16, 17, 18, 19]
08-24 17:04:32.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [20, 21, 22, 23, 24]
08-24 17:04:37.162 23975-24014/com.example.wty.learnrxjava I/MainActivity: accept: longs = [25, 26, 27, 28, 29]

• GroupBy
  将一个Observable分拆为一些Observables集合，它们中的每一个发射原始Observable的一个子序列

groupBy.c.png

GroupBy操作符将原始Observable分拆为一些Observables集合，它们中的每一个发射原始Observable数据序列的一个子序列。哪个数据项由哪一个Observable发射是由一个函数判定的，这个函数给每一项指定一个Key，Key相同的数据会被同一个Observable发射。
举个例子,

Observable.interval(1, TimeUnit.SECONDS)
                .take(30)
                .groupBy(new Function<Long, Integer>() {
                    @Override
                    public Integer apply(@NonNull Long aLong) throws Exception {
                        if (aLong % 3 == 0) {
                            return 3;
                        } else if (aLong % 4 == 0) {
                            return 4;
                        } else if (aLong % 5 == 0) {
                            return 5;
                        } else {
                            return 1;
                        }

                    }
                })
                .subscribe(new Consumer<GroupedObservable<Integer, Long>>() {
                    @Override
                    public void accept(GroupedObservable<Integer, Long> longLongGroupedObservable) throws Exception {
                        Integer key = longLongGroupedObservable.getKey();
                        Log.i(TAG, "accept: key = " + key);
                        if (key == 3) {
                            longLongGroupedObservable.subscribe(new Consumer<Long>() {
                                @Override
                                public void accept(Long aLong) throws Exception {
                                    Log.i(TAG, "accept: key = 3,3的倍数,aLong = " + aLong);
                                }
                            });
                        } else if (key == 4) {
                            longLongGroupedObservable.subscribe(new Consumer<Long>() {
                                @Override
                                public void accept(Long aLong) throws Exception {
                                    Log.i(TAG, "accept: key = 4,4的倍数 aLong = " + aLong);
                                }
                            });
                        } else if (key == 5) {
                            longLongGroupedObservable.subscribe(new Consumer<Long>() {
                                @Override
                                public void accept(Long aLong) throws Exception {
                                    Log.i(TAG, "accept: key = 5,5的倍数 aLong = " + aLong);
                                }
                            });
                        } else {
                            longLongGroupedObservable.subscribe(new Consumer<Long>() {
                                @Override
                                public void accept(Long aLong) throws Exception {
                                    Log.i(TAG, "accept: key = 1,不是3、4、5的倍数 aLong = " + aLong);
                                }
                            });
                        }
                    }
                });

08-24 17:22:37.134 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1
08-24 17:22:37.135 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 1
08-24 17:22:37.231 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 2
08-24 17:22:37.331 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3
08-24 17:22:37.332 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 3
08-24 17:22:37.431 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4
08-24 17:22:37.432 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 4
08-24 17:22:37.531 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 5
08-24 17:22:37.532 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 5,5的倍数 aLong = 5
08-24 17:22:37.631 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 6
08-24 17:22:37.731 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 7
08-24 17:22:37.831 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 8
08-24 17:22:37.931 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 9
08-24 17:22:38.031 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 5,5的倍数 aLong = 10
08-24 17:22:38.131 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 11
08-24 17:22:38.231 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 12
08-24 17:22:38.331 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 13
08-24 17:22:38.431 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 14
08-24 17:22:38.531 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 15
08-24 17:22:38.631 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 16
08-24 17:22:38.731 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 17
08-24 17:22:38.831 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 18
08-24 17:22:38.931 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 19
08-24 17:22:39.031 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 20
08-24 17:22:39.131 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 21
08-24 17:22:39.231 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 22
08-24 17:22:39.331 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 23
08-24 17:22:39.431 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 24
08-24 17:22:39.531 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 5,5的倍数 aLong = 25
08-24 17:22:39.631 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 26
08-24 17:22:39.731 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 27
08-24 17:22:39.831 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 4,4的倍数 aLong = 28
08-24 17:22:39.932 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 1,不是3、4、5的倍数 aLong = 29
08-24 17:22:40.031 7842-7881/com.example.wty.learnrxjava I/MainActivity: accept: key = 3,3的倍数,aLong = 30

过滤操作

• Distinct
  抑制（过滤掉）重复的数据项

distinct.png

Observable.just(1,2,1,1,2,1,1,2,3,4)
                .distinct(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply(@NonNull Integer integer) throws Exception {
                        return integer;
                    }
                })
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "accept: integer = " + integer);
                    }
                });

08-24 17:26:19.198 10816-10816/com.example.wty.learnrxjava I/MainActivity: accept: integer = 1
08-24 17:26:19.199 10816-10816/com.example.wty.learnrxjava I/MainActivity: accept: integer = 2
08-24 17:26:19.199 10816-10816/com.example.wty.learnrxjava I/MainActivity: accept: integer = 3
08-24 17:26:19.199 10816-10816/com.example.wty.learnrxjava I/MainActivity: accept: integer = 4

• ElementAt
  只发射第N项数据
• Filter
  Filter操作符使用你指定的一个谓词函数测试数据项，只有通过测试的数据才会被发射。
  举个例子,过滤整数序列中的奇数,只发射偶数
  filter.png

Observable.interval(1,TimeUnit.SECONDS)
                .filter(new Predicate<Long>() {
                    @Override
                    public boolean test(@NonNull Long aLong) throws Exception {
                        if (aLong % 2 == 0){
                            return true;
                        }
                        return false;
                    }
                })
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        Log.i(TAG, "accept: " + aLong);
                    }
                });

08-24 17:30:57.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 0
08-24 17:30:59.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 2
08-24 17:31:01.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 4
08-24 17:31:03.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 6
08-24 17:31:05.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 8
08-24 17:31:07.613 14530-14549/com.example.wty.learnrxjava I/MainActivity: accept: 10

• First
  只发射第一项（或者满足某个条件的第一项）数据
• IgnoreElements
  不发射任何数据，只发射Observable的终止通知
• skip/take
  skip
  image.png
  take
  image.png

• skipLast/takeLast

image.png image.png

• Sample
  定期发射Observable最近发射的数据项

image.png

RxJava将这个操作符实现为sample和throttleLast。

Observable.range(0,1000)
                .sample(1, TimeUnit.MICROSECONDS)
                .subscribe(new Consumer<Integer>() {
                    @Override
                    public void accept(Integer integer) throws Exception {
                        Log.i(TAG, "accept: " + integer);
                    }
                });

08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 56
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 154
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 186
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 208
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 228
08-24 17:41:23.376 22979-23017/com.example.wty.learnrxjava I/MainActivity: accept: 247

组合操作

• zip
  通过一个函数将多个Observables的发射物结合到一起，基于这个函数的结果为每个结合体发射单个数据项。
  Zip操作符返回一个Obversable，它使用这个函数按顺序结合两个或多个Observables发射的数据项，然后它发射这个函数返回的结果。它按照严格的顺序应用这个函数。它只发射与发射数据项最少的那个Observable一样多的数据。

image.png

上代码:

Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                Thread.sleep(1000);
                e.onNext(1);
                Thread.sleep(1000);
                e.onNext(2);
                Thread.sleep(1000);
                e.onNext(3);
                Thread.sleep(1000);
                e.onNext(4);
                Thread.sleep(1000);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.newThread());

        Observable<String> stringObservable = Observable.create(new ObservableOnSubscribe<String>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                Thread.sleep(700);
                e.onNext("A");
                Thread.sleep(700);
                e.onNext("B");
                Thread.sleep(700);
                e.onNext("C");
                Thread.sleep(700);
                e.onNext("D");
                Thread.sleep(700);
                e.onNext("E");
                Thread.sleep(700);
                e.onNext("F");
                Thread.sleep(700);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.newThread());
        Observable
                .zip(stringObservable, integerObservable, new BiFunction<String, Integer, String>() {
                    @Override
                    public String apply(@NonNull String s, @NonNull Integer integer) throws Exception {
                        return s + integer;
                    }
                })
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(@NonNull Disposable d) {
                        Log.i(TAG, "onSubscribe: ");
                    }

                    @Override
                    public void onNext(@NonNull String s) {
                        Log.i(TAG, "onNext: " +s);
                    }

                    @Override
                    public void onError(@NonNull Throwable e) {

                    }

                    @Override
                    public void onComplete() {
                        Log.i(TAG, "onComplete: ");
                    }
                });

08-24 17:50:11.195 30117-30157/com.example.wty.learnrxjava I/MainActivity: onNext: A1
08-24 17:50:12.196 30117-30157/com.example.wty.learnrxjava I/MainActivity: onNext: B2
08-24 17:50:13.197 30117-30157/com.example.wty.learnrxjava I/MainActivity: onNext: C3
08-24 17:50:14.198 30117-30157/com.example.wty.learnrxjava I/MainActivity: onNext: D4
08-24 17:50:15.198 30117-30157/com.example.wty.learnrxjava I/MainActivity: onComplete: 

错误处理

• Retry
  如果原始Observable遇到错误(即接收到onError的时候触发)，重新订阅它期望它能正常终止

image.png

• RetryWhen
  retryWhen和retry类似，区别是，retryWhen将onError中的Throwable传递给一个函数，这个函数产生另一个Observable，retryWhen观察它的结果再决定是不是要重新订阅原始的Observable。如果这个Observable发射了一项数据，它就重新订阅，如果这个Observable发射的是onError通知，它就将这个通知传递给观察者然后终止。

 Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(@NonNull ObservableEmitter<Integer> e) throws Exception {
                Thread.sleep(1000);
                e.onNext(1);
                Thread.sleep(1000);
                e.onNext(2);
                Thread.sleep(1000);
                e.onNext(3);
                int value = 1 / 0;
                Thread.sleep(1000);
                e.onNext(4);
                Thread.sleep(1000);
                e.onComplete();
            }
        }).subscribeOn(Schedulers.newThread());
integerObservable.retryWhen(new Function<Observable<Throwable>, ObservableSource<?>>() {
            @Override
            public ObservableSource<?> apply(@NonNull Observable<Throwable> throwableObservable) throws Exception {

                return throwableObservable.flatMap(new Function<Throwable, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(@NonNull Throwable throwable) throws Exception {
                        if (throwable instanceof  ArithmeticException)
                            return Observable.just(1);
                        else
                            return Observable.error(throwable);
                    }
                });
            }
        }).subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(@NonNull Disposable d) {
                Log.i(TAG, "onSubscribe: ");
            }

            @Override
            public void onNext(@NonNull Integer integer) {
                Log.i(TAG, "onNext: " + integer);
            }

            @Override
            public void onError(@NonNull Throwable e) {
                Log.e(TAG, "onError: ", e);
            }

            @Override
            public void onComplete() {
                Log.i(TAG, "onComplete: ");
            }
        });

08-24 17:56:24.620 3895-3935/com.example.wty.learnrxjava I/MainActivity: onNext: 1
08-24 17:56:25.621 3895-3935/com.example.wty.learnrxjava I/MainActivity: onNext: 2
08-24 17:56:26.621 3895-3935/com.example.wty.learnrxjava I/MainActivity: onNext: 3
08-24 17:56:27.624 3895-3998/com.example.wty.learnrxjava I/MainActivity: onNext: 1
08-24 17:56:28.624 3895-3998/com.example.wty.learnrxjava I/MainActivity: onNext: 2
08-24 17:56:29.625 3895-3998/com.example.wty.learnrxjava I/MainActivity: onNext: 3
08-24 17:56:30.629 3895-4038/com.example.wty.learnrxjava I/MainActivity: onNext: 1
08-24 17:56:31.630 3895-4038/com.example.wty.learnrxjava I/MainActivity: onNext: 2
08-24 17:56:32.630 3895-4038/com.example.wty.learnrxjava I/MainActivity: onNext: 3

线程调度

• SubscribeOn
  指定Observable自身在哪个调度器上执行
• ObserveOn
  指定一个观察者在哪个调度器上观察这个Observable