Merge branch 'master' into 3.0.0
Nikita Koksharov
6 years ago
commit
4abd81e49f
@ -1,5 +1,7 @@
|
||||
## Redisson standalone node
|
||||
|
||||
Redisson offers ability to run as standalone node and participate in distributed computing. Such standalone nodes could be used to run [ExecutorService](./9.-distributed-services#93-executor-service), [ScheduledExecutorService](https://github.com/mrniko/redisson/wiki/9.-distributed-services#94-scheduled-executor-service) tasks or [RemoteService](./9.-distributed-services#91-remote-service) services. It's just a single jar and could be downloaded from [here](https://repository.sonatype.org/service/local/artifact/maven/redirect?r=central-proxy&g=org.redisson&a=redisson-all&v=2.3.0&e=jar)
|
||||
Redisson offers ability to run as standalone node and participate in distributed computing. Such Nodes are used to run [MapReduce](./9.-distributed-services/#95-distributed-mapreduce-service), [ExecutorService](./9.-distributed-services#93-executor-service), [ScheduledExecutorService](https://github.com/mrniko/redisson/wiki/9.-distributed-services#94-scheduled-executor-service) tasks or [RemoteService](./9.-distributed-services#91-remote-service) services. All tasks are kept in Redis until their execution moment.
|
||||
|
||||
Packaged as a single jar and could be [downloaded](https://repository.sonatype.org/service/local/artifact/maven/redirect?r=central-proxy&g=org.redisson&a=redisson-all&v=3.9.0&e=jar).
|
||||
|
||||
[Documentation](https://github.com/mrniko/redisson/wiki/12.-Standalone-node) about Redisson standalone node.
|
||||
|
||||
@ -0,0 +1,62 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.
|
||||
*/
|
||||
package org.redisson.client.protocol.decoder;
|
||||
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.redisson.client.handler.State;
|
||||
import org.redisson.client.protocol.Decoder;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class StreamObjectMapReplayDecoder extends ObjectMapReplayDecoder {
|
||||
|
||||
private Decoder<Object> codec;
|
||||
|
||||
public StreamObjectMapReplayDecoder() {
|
||||
}
|
||||
|
||||
public StreamObjectMapReplayDecoder(Decoder<Object> codec) {
|
||||
super();
|
||||
this.codec = codec;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<Object, Object> decode(List<Object> parts, State state) {
|
||||
if (parts.get(0) instanceof Map) {
|
||||
Map<Object, Object> result = new LinkedHashMap<Object, Object>(parts.size());
|
||||
for (int i = 0; i < parts.size(); i++) {
|
||||
result.putAll((Map<? extends Object, ? extends Object>) parts.get(i));
|
||||
}
|
||||
return result;
|
||||
}
|
||||
return super.decode(parts, state);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Decoder<Object> getDecoder(int paramNum, State state) {
|
||||
if (codec != null) {
|
||||
return codec;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,79 @@
|
||||
/**
|
||||
* Copyright 2018 Nikita Koksharov
|
||||
*
|
||||
* 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.
|
||||
*/
|
||||
package org.redisson.rx;
|
||||
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
import java.util.concurrent.atomic.AtomicReference;
|
||||
|
||||
import org.redisson.api.RFuture;
|
||||
|
||||
import io.netty.util.concurrent.Future;
|
||||
import io.netty.util.concurrent.FutureListener;
|
||||
import io.reactivex.Flowable;
|
||||
import io.reactivex.functions.Action;
|
||||
import io.reactivex.functions.LongConsumer;
|
||||
import io.reactivex.processors.ReplayProcessor;
|
||||
|
||||
/**
|
||||
*
|
||||
* @author Nikita Koksharov
|
||||
*
|
||||
*/
|
||||
public class ElementsStream {
|
||||
|
||||
public static <V> Flowable<V> takeElements(final Callable<RFuture<V>> callable) {
|
||||
final ReplayProcessor<V> p = ReplayProcessor.create();
|
||||
return p.doOnRequest(new LongConsumer() {
|
||||
@Override
|
||||
public void accept(long n) throws Exception {
|
||||
final AtomicLong counter = new AtomicLong(n);
|
||||
final AtomicReference<RFuture<V>> futureRef = new AtomicReference<RFuture<V>>();
|
||||
|
||||
take(callable, p, counter, futureRef);
|
||||
|
||||
p.doOnCancel(new Action() {
|
||||
@Override
|
||||
public void run() throws Exception {
|
||||
futureRef.get().cancel(true);
|
||||
}
|
||||
});
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private static <V> void take(final Callable<RFuture<V>> factory, final ReplayProcessor<V> p, final AtomicLong counter, final AtomicReference<RFuture<V>> futureRef) throws Exception {
|
||||
RFuture<V> future = factory.call();
|
||||
futureRef.set(future);
|
||||
future.addListener(new FutureListener<V>() {
|
||||
@Override
|
||||
public void operationComplete(Future<V> future) throws Exception {
|
||||
if (!future.isSuccess()) {
|
||||
p.onError(future.cause());
|
||||
return;
|
||||
}
|
||||
|
||||
p.onNext(future.getNow());
|
||||
if (counter.decrementAndGet() == 0) {
|
||||
p.onComplete();
|
||||
}
|
||||
|
||||
take(factory, p, counter, futureRef);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue