RedisProtocol extends AbstractProtocol { @Override public <T> Invoker<T> refer(final Class<T> type, final URL url) throws RpcException { try { GenericObjectPoolConfig config = new GenericObjectPoolConfig(); config.setTestOnBorrow(url.getParameter("test.on.borrow", true)); config.setTestOnReturn(url.getParameter("test.on.return", false)); config.setTestWhileIdle(url.getParameter("test.while.idle", false)); if (url.getParameter("max.idle", 0) > 0) config.setMaxIdle(url.getParameter("max.idle", 0)); if (url.getParameter("min.idle", 0) > 0) config.setMinIdle(url.getParameter("min.idle", 0)); if (url.getParameter("max.active", 0) > 0) config.setMaxTotal(url.getParameter("max.active", 0)); if (url.getParameter("max.total", 0) > 0) config.setMaxTotal(url.getParameter("max.total", 0)); if (url.getParameter("max.wait", 0) > 0) config.setMaxWaitMillis(url.getParameter("max.wait", 0)); if (url.getParameter("num.tests.per.eviction.run", 0) > 0) config.setNumTestsPerEvictionRun(url.getParameter("num.tests.per.eviction.run", 0)); if (url.getParameter("time.between.eviction.runs.millis", 0) > 0) config.setTimeBetweenEvictionRunsMillis(url.getParameter("time.between.eviction.runs.millis", 0)); if (url.getParameter("min.evictable.idle.time.millis", 0) > 0) config.setMinEvictableIdleTimeMillis(url.getParameter("min.evictable.idle.time.millis", 0)); final JedisPool jedisPool = new JedisPool(config, url.getHost(), url.getPort(DEFAULT_PORT), url.getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT), StringUtils.isBlank(url.getPassword()) ? null : url.getPassword(), url.getParameter("db.index", 0)); final int expiry = url.getParameter("expiry", 0); final String get = url.getParameter("get", "get"); final String set = url.getParameter("set", Map.class.equals(type) ? "put" : "set"); final String delete = url.getParameter("delete", Map.class.equals(type) ? "remove" : "delete"); return new AbstractInvoker<T>(type, url) { @Override protected Result doInvoke(Invocation invocation) throws Throwable { Jedis resource = null; try { resource = jedisPool.getResource(); if (get.equals(invocation.getMethodName())) { if (invocation.getArguments().length != 1) { throw new IllegalArgumentException("The redis get method arguments mismatch, must only one arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url); } byte[] value = resource.get(String.valueOf(invocation.getArguments()[0]).getBytes()); if (value == null) { return new RpcResult(); } ObjectInput oin = getSerialization(url).deserialize(url, new ByteArrayInputStream(value)); return new RpcResult(oin.readObject()); } else if (set.equals(invocation.getMethodName())) { if (invocation.getArguments().length != 2) { throw new IllegalArgumentException("The redis set method arguments mismatch, must be two arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url); } byte[] key = String.valueOf(invocation.getArguments()[0]).getBytes(); ByteArrayOutputStream output = new ByteArrayOutputStream(); ObjectOutput value = getSerialization(url).serialize(url, output); value.writeObject(invocation.getArguments()[1]); resource.set(key, output.toByteArray()); if (expiry > 1000) { resource.expire(key, expiry / 1000); } return new RpcResult(); } else if (delete.equals(invocation.getMethodName())) { if (invocation.getArguments().length != 1) { throw new IllegalArgumentException("The redis delete method arguments mismatch, must only one arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url); } resource.del(String.valueOf(invocation.getArguments()[0]).getBytes()); return new RpcResult(); } else { throw new UnsupportedOperationException("Unsupported method " + invocation.getMethodName() + " in redis service."); } } catch (Throwable t) { RpcException re = new RpcException("Failed to invoke redis service method. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url + ", cause: " + t.getMessage(), t); if (t instanceof TimeoutException || t instanceof SocketTimeoutException) { re.setCode(RpcException.TIMEOUT_EXCEPTION); } else if (t instanceof JedisConnectionException || t instanceof IOException) { re.setCode(RpcException.NETWORK_EXCEPTION); } else if (t instanceof JedisDataException) { re.setCode(RpcException.SERIALIZATION_EXCEPTION); } throw re; } finally { if (resource != null) { try { jedisPool.returnResource(resource); } catch (Throwable t) { logger.warn("returnResource error: " + t.getMessage(), t); } } } } @Override public void destroy() { super.destroy(); try { jedisPool.destroy(); } catch (Throwable e) { logger.warn(e.getMessage(), e); } } }; } catch (Throwable t) { throw new RpcException("Failed to refer redis service. interface: " + type.getName() + ", url: " + url + ", cause: " + t.getMessage(), t); } }  @Override int getDefaultPort(); @Override Exporter<T> export(final Invoker<T> invoker); @Override Invoker<T> refer(final Class<T> type, final URL url); static final int DEFAULT_PORT; }
@Test(expected = RpcException.class) public void testUnsupportedMethod() { Invoker<IDemoService> refer = protocol.refer(IDemoService.class, registryUrl); IDemoService demoService = this.proxy.getProxy(refer); demoService.unsupported(null); }
@Test(expected = RpcException.class) public void testWrongParameters() { Invoker<IDemoService> refer = protocol.refer(IDemoService.class, registryUrl); IDemoService demoService = this.proxy.getProxy(refer); demoService.set("key", "value", "wrongValue"); }
@Test(expected = RpcException.class) public void testWrongRedis() { Invoker<IDemoService> refer = protocol.refer(IDemoService.class, URL.valueOf("redis: IDemoService demoService = this.proxy.getProxy(refer); demoService.get("key"); }
@Test public void testAuthRedis() { Invoker<IDemoService> refer = protocol.refer(IDemoService.class, registryUrl .addParameter("max.idle", 10) .addParameter("max.active", 20)); IDemoService demoService = this.proxy.getProxy(refer); String value = demoService.get("key"); assertThat(value, is(nullValue())); demoService.set("key", "newValue"); value = demoService.get("key"); assertThat(value, is("newValue")); demoService.delete("key"); value = demoService.get("key"); assertThat(value, is(nullValue())); refer.destroy(); String password = "123456"; int database = 1; this.registryUrl = this.registryUrl.setPassword(password).addParameter("db.index", database); refer = protocol.refer(IDemoService.class, registryUrl .addParameter("max.idle", 10) .addParameter("max.active", 20)); demoService = this.proxy.getProxy(refer); demoService.set("key", "newValue"); value = demoService.get("key"); assertThat(value, is("newValue")); JedisPool pool = new JedisPool(new GenericObjectPoolConfig(), "localhost", registryUrl.getPort(), 2000, password, database, (String)null); Jedis jedis = null; try { jedis = pool.getResource(); byte[] valueByte = jedis.get("key".getBytes()); Serialization serialization = ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(this.registryUrl.getParameter(Constants.SERIALIZATION_KEY, "java")); ObjectInput oin = serialization.deserialize(this.registryUrl, new ByteArrayInputStream(valueByte)); String actual = (String) oin.readObject(); assertThat(value, is(actual)); } catch(Exception e) { Assert.fail("jedis gets the result comparison is error!"); } finally { if (jedis != null) { jedis.close(); } pool.destroy(); } demoService.delete("key"); value = demoService.get("key"); assertThat(value, is(nullValue())); refer.destroy(); }
@Test public void testWrongAuthRedis() { String password = "1234567"; this.registryUrl = this.registryUrl.setPassword(password); Invoker<IDemoService> refer = protocol.refer(IDemoService.class, registryUrl .addParameter("max.idle", 10) .addParameter("max.active", 20)); IDemoService demoService = this.proxy.getProxy(refer); try { String value = demoService.get("key"); assertThat(value, is(nullValue())); } catch (RpcException e) { if (e.getCause() instanceof JedisConnectionException && e.getCause().getCause() instanceof JedisDataException) { Assert.assertEquals("ERR invalid password" , e.getCause().getCause().getMessage()); } else { Assert.fail("no invalid password exception!"); } } refer.destroy(); }
@Test public void testReferClass() { Invoker<IDemoService> refer = protocol.refer(IDemoService.class, registryUrl); Class<IDemoService> serviceClass = refer.getInterface(); assertThat(serviceClass.getName(), is("com.alibaba.dubbo.rpc.protocol.redis.IDemoService")); }
@Test public void testInvocation() { Invoker<IDemoService> refer = protocol.refer(IDemoService.class, registryUrl .addParameter("max.idle", 10) .addParameter("max.active", 20)); IDemoService demoService = this.proxy.getProxy(refer); String value = demoService.get("key"); assertThat(value, is(nullValue())); demoService.set("key", "newValue"); value = demoService.get("key"); assertThat(value, is("newValue")); demoService.delete("key"); value = demoService.get("key"); assertThat(value, is(nullValue())); refer.destroy(); }