@Override
            public Observable<Void> call(ByteBuf byteBuf) {
                String movieId = byteBuf.toString(Charset.defaultCharset());
                System.out.println(format("    updating: {user=%s, movie=%s}", userId, movieId));
                synchronized (this) {
                    Set<String> recommendations;
                    if (userRecommendations.containsKey(userId)) {
                        recommendations = userRecommendations.get(userId);
                    } else {
                        recommendations = new ConcurrentSet<String>();
                        userRecommendations.put(userId, recommendations);
                    }
                    recommendations.add(movieId);
                }
                response.setStatus(HttpResponseStatus.OK);
                return response.close();
            }
The client examples assume that the movie server runs on the default port 8080.

@author Tomasz Bak
ribbon-examples/src/main/java/com/netflix/ribbon/examples/rx/RxMovieServer.java