mirror of https://github.com/apache/cassandra
updated subversion properties (see: CASSANDRA-429)
git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/branches/cassandra-0.4@812260 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Eric Evans
parent
7ec7106b70
commit
8070b9bdf0
|
|
@ -1,107 +1,107 @@
|
|||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you 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.apache.cassandra.client;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.dht.IPartitioner;
|
||||
import org.apache.cassandra.dht.Token;
|
||||
import org.apache.cassandra.locator.AbstractReplicationStrategy;
|
||||
import org.apache.cassandra.locator.TokenMetadata;
|
||||
import org.apache.cassandra.net.EndPoint;
|
||||
import org.apache.cassandra.service.Cassandra;
|
||||
import org.apache.cassandra.service.CassandraServer;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.log4j.Logger;
|
||||
import org.apache.thrift.TException;
|
||||
import org.apache.thrift.protocol.TBinaryProtocol;
|
||||
import org.apache.thrift.transport.TSocket;
|
||||
|
||||
import flexjson.JSONTokener;
|
||||
|
||||
/**
|
||||
* A class for caching the ring map at the client. For usage example, see
|
||||
* test/unit/org.apache.cassandra.client.TestRingCache.java.
|
||||
*/
|
||||
public class RingCache
|
||||
{
|
||||
final private static Logger logger_ = Logger.getLogger(RingCache.class);
|
||||
|
||||
private Set<String> seeds_ = new HashSet<String>();
|
||||
final private int port_=DatabaseDescriptor.getThriftPort();
|
||||
private volatile AbstractReplicationStrategy nodePicker_;
|
||||
final private static IPartitioner partitioner_ = DatabaseDescriptor.getPartitioner();
|
||||
|
||||
public RingCache()
|
||||
{
|
||||
seeds_ = DatabaseDescriptor.getSeeds();
|
||||
refreshEndPointMap();
|
||||
}
|
||||
|
||||
public void refreshEndPointMap()
|
||||
{
|
||||
for (String seed : seeds_)
|
||||
{
|
||||
try
|
||||
{
|
||||
TSocket socket = new TSocket(seed, port_);
|
||||
TBinaryProtocol binaryProtocol = new TBinaryProtocol(socket, false, false);
|
||||
Cassandra.Client client = new Cassandra.Client(binaryProtocol);
|
||||
socket.open();
|
||||
|
||||
Map<String,String> tokenToHostMap = (Map<String,String>) new JSONTokener(client.get_string_property(CassandraServer.TOKEN_MAP)).nextValue();
|
||||
|
||||
HashMap<Token, EndPoint> tokenEndpointMap = new HashMap<Token, EndPoint>();
|
||||
Map<EndPoint, Token> endpointTokenMap = new HashMap<EndPoint, Token>();
|
||||
for (Map.Entry<String,String> entry : tokenToHostMap.entrySet())
|
||||
{
|
||||
Token token = StorageService.getPartitioner().getTokenFactory().fromString(entry.getKey());
|
||||
String host = entry.getValue();
|
||||
tokenEndpointMap.put(token, new EndPoint(host, port_));
|
||||
endpointTokenMap.put(new EndPoint(host, port_), token);
|
||||
}
|
||||
|
||||
TokenMetadata tokenMetadata = new TokenMetadata(tokenEndpointMap, endpointTokenMap, null);
|
||||
Class cls = DatabaseDescriptor.getReplicaPlacementStrategyClass();
|
||||
Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class, int.class};
|
||||
try
|
||||
{
|
||||
nodePicker_ = (AbstractReplicationStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata, partitioner_, DatabaseDescriptor.getReplicationFactor(), port_);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
break;
|
||||
}
|
||||
catch (TException e)
|
||||
{
|
||||
/* let the Exception go and try another seed. log this though */
|
||||
logger_.debug("Error contacting seed " + seed + " " + e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public EndPoint[] getEndPoint(String key)
|
||||
{
|
||||
return nodePicker_.getReadStorageEndPoints(partitioner_.getToken(key));
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you 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.apache.cassandra.client;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import org.apache.cassandra.config.DatabaseDescriptor;
|
||||
import org.apache.cassandra.dht.IPartitioner;
|
||||
import org.apache.cassandra.dht.Token;
|
||||
import org.apache.cassandra.locator.AbstractReplicationStrategy;
|
||||
import org.apache.cassandra.locator.TokenMetadata;
|
||||
import org.apache.cassandra.net.EndPoint;
|
||||
import org.apache.cassandra.service.Cassandra;
|
||||
import org.apache.cassandra.service.CassandraServer;
|
||||
import org.apache.cassandra.service.StorageService;
|
||||
import org.apache.log4j.Logger;
|
||||
import org.apache.thrift.TException;
|
||||
import org.apache.thrift.protocol.TBinaryProtocol;
|
||||
import org.apache.thrift.transport.TSocket;
|
||||
|
||||
import flexjson.JSONTokener;
|
||||
|
||||
/**
|
||||
* A class for caching the ring map at the client. For usage example, see
|
||||
* test/unit/org.apache.cassandra.client.TestRingCache.java.
|
||||
*/
|
||||
public class RingCache
|
||||
{
|
||||
final private static Logger logger_ = Logger.getLogger(RingCache.class);
|
||||
|
||||
private Set<String> seeds_ = new HashSet<String>();
|
||||
final private int port_=DatabaseDescriptor.getThriftPort();
|
||||
private volatile AbstractReplicationStrategy nodePicker_;
|
||||
final private static IPartitioner partitioner_ = DatabaseDescriptor.getPartitioner();
|
||||
|
||||
public RingCache()
|
||||
{
|
||||
seeds_ = DatabaseDescriptor.getSeeds();
|
||||
refreshEndPointMap();
|
||||
}
|
||||
|
||||
public void refreshEndPointMap()
|
||||
{
|
||||
for (String seed : seeds_)
|
||||
{
|
||||
try
|
||||
{
|
||||
TSocket socket = new TSocket(seed, port_);
|
||||
TBinaryProtocol binaryProtocol = new TBinaryProtocol(socket, false, false);
|
||||
Cassandra.Client client = new Cassandra.Client(binaryProtocol);
|
||||
socket.open();
|
||||
|
||||
Map<String,String> tokenToHostMap = (Map<String,String>) new JSONTokener(client.get_string_property(CassandraServer.TOKEN_MAP)).nextValue();
|
||||
|
||||
HashMap<Token, EndPoint> tokenEndpointMap = new HashMap<Token, EndPoint>();
|
||||
Map<EndPoint, Token> endpointTokenMap = new HashMap<EndPoint, Token>();
|
||||
for (Map.Entry<String,String> entry : tokenToHostMap.entrySet())
|
||||
{
|
||||
Token token = StorageService.getPartitioner().getTokenFactory().fromString(entry.getKey());
|
||||
String host = entry.getValue();
|
||||
tokenEndpointMap.put(token, new EndPoint(host, port_));
|
||||
endpointTokenMap.put(new EndPoint(host, port_), token);
|
||||
}
|
||||
|
||||
TokenMetadata tokenMetadata = new TokenMetadata(tokenEndpointMap, endpointTokenMap, null);
|
||||
Class cls = DatabaseDescriptor.getReplicaPlacementStrategyClass();
|
||||
Class [] parameterTypes = new Class[] { TokenMetadata.class, IPartitioner.class, int.class, int.class};
|
||||
try
|
||||
{
|
||||
nodePicker_ = (AbstractReplicationStrategy) cls.getConstructor(parameterTypes).newInstance(tokenMetadata, partitioner_, DatabaseDescriptor.getReplicationFactor(), port_);
|
||||
}
|
||||
catch (Exception e)
|
||||
{
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
break;
|
||||
}
|
||||
catch (TException e)
|
||||
{
|
||||
/* let the Exception go and try another seed. log this though */
|
||||
logger_.debug("Error contacting seed " + seed + " " + e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public EndPoint[] getEndPoint(String key)
|
||||
{
|
||||
return nodePicker_.getReadStorageEndPoints(partitioner_.getToken(key));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,81 +1,81 @@
|
|||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you 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.apache.cassandra.client;
|
||||
|
||||
import org.apache.cassandra.net.EndPoint;
|
||||
import org.apache.cassandra.service.Cassandra;
|
||||
import org.apache.cassandra.service.Column;
|
||||
import org.apache.cassandra.service.ColumnPath;
|
||||
import org.apache.thrift.protocol.TBinaryProtocol;
|
||||
import org.apache.thrift.transport.TSocket;
|
||||
import org.apache.thrift.transport.TTransport;
|
||||
|
||||
/**
|
||||
* Sample code that uses RingCache in the client.
|
||||
*/
|
||||
public class TestRingCache
|
||||
{
|
||||
private static RingCache ringCache;
|
||||
private static Cassandra.Client thriftClient;
|
||||
|
||||
static
|
||||
{
|
||||
ringCache = new RingCache();
|
||||
}
|
||||
|
||||
private static void setup(String server, int port) throws Exception
|
||||
{
|
||||
/* Establish a thrift connection to the cassandra instance */
|
||||
TSocket socket = new TSocket(server, port);
|
||||
TTransport transport;
|
||||
System.out.println(" connected to " + server + ":" + port + ".");
|
||||
transport = socket;
|
||||
TBinaryProtocol binaryProtocol = new TBinaryProtocol(transport, false, false);
|
||||
Cassandra.Client cassandraClient = new Cassandra.Client(binaryProtocol);
|
||||
transport.open();
|
||||
thriftClient = cassandraClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* usage: java -Dstorage-config="confpath" org.apache.cassandra.client.TestRingCache
|
||||
* @param args
|
||||
* @throws Exception
|
||||
*/
|
||||
public static void main(String[] args) throws Throwable
|
||||
{
|
||||
String table = "Keyspace1";
|
||||
for (int nRows=1; nRows<10; nRows++)
|
||||
{
|
||||
String row = "row" + nRows;
|
||||
ColumnPath col = new ColumnPath("Standard1", null, "col1".getBytes());
|
||||
|
||||
EndPoint endPoints[] = ringCache.getEndPoint(row);
|
||||
String hosts="";
|
||||
for (int i=0; i<endPoints.length; i++)
|
||||
hosts = hosts + ((i>0) ? "," : "") + endPoints[i].getHost();
|
||||
System.out.println("hosts with key " + row + " : " + hosts + "; choose " + endPoints[0].getHost());
|
||||
|
||||
// now, read the row back directly from the host owning the row locally
|
||||
setup(endPoints[0].getHost(), endPoints[0].getPort());
|
||||
thriftClient.insert(table, row, col, "val1".getBytes(), 1, 1);
|
||||
Column column=thriftClient.get(table, row, col, 1).column;
|
||||
System.out.println("read row " + row + " " + new String(column.name) + ":" + new String(column.value) + ":" + column.timestamp);
|
||||
}
|
||||
System.exit(1);
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you 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.apache.cassandra.client;
|
||||
|
||||
import org.apache.cassandra.net.EndPoint;
|
||||
import org.apache.cassandra.service.Cassandra;
|
||||
import org.apache.cassandra.service.Column;
|
||||
import org.apache.cassandra.service.ColumnPath;
|
||||
import org.apache.thrift.protocol.TBinaryProtocol;
|
||||
import org.apache.thrift.transport.TSocket;
|
||||
import org.apache.thrift.transport.TTransport;
|
||||
|
||||
/**
|
||||
* Sample code that uses RingCache in the client.
|
||||
*/
|
||||
public class TestRingCache
|
||||
{
|
||||
private static RingCache ringCache;
|
||||
private static Cassandra.Client thriftClient;
|
||||
|
||||
static
|
||||
{
|
||||
ringCache = new RingCache();
|
||||
}
|
||||
|
||||
private static void setup(String server, int port) throws Exception
|
||||
{
|
||||
/* Establish a thrift connection to the cassandra instance */
|
||||
TSocket socket = new TSocket(server, port);
|
||||
TTransport transport;
|
||||
System.out.println(" connected to " + server + ":" + port + ".");
|
||||
transport = socket;
|
||||
TBinaryProtocol binaryProtocol = new TBinaryProtocol(transport, false, false);
|
||||
Cassandra.Client cassandraClient = new Cassandra.Client(binaryProtocol);
|
||||
transport.open();
|
||||
thriftClient = cassandraClient;
|
||||
}
|
||||
|
||||
/**
|
||||
* usage: java -Dstorage-config="confpath" org.apache.cassandra.client.TestRingCache
|
||||
* @param args
|
||||
* @throws Exception
|
||||
*/
|
||||
public static void main(String[] args) throws Throwable
|
||||
{
|
||||
String table = "Keyspace1";
|
||||
for (int nRows=1; nRows<10; nRows++)
|
||||
{
|
||||
String row = "row" + nRows;
|
||||
ColumnPath col = new ColumnPath("Standard1", null, "col1".getBytes());
|
||||
|
||||
EndPoint endPoints[] = ringCache.getEndPoint(row);
|
||||
String hosts="";
|
||||
for (int i=0; i<endPoints.length; i++)
|
||||
hosts = hosts + ((i>0) ? "," : "") + endPoints[i].getHost();
|
||||
System.out.println("hosts with key " + row + " : " + hosts + "; choose " + endPoints[0].getHost());
|
||||
|
||||
// now, read the row back directly from the host owning the row locally
|
||||
setup(endPoints[0].getHost(), endPoints[0].getPort());
|
||||
thriftClient.insert(table, row, col, "val1".getBytes(), 1, 1);
|
||||
Column column=thriftClient.get(table, row, col, 1).column;
|
||||
System.out.println("read row " + row + " " + new String(column.name) + ":" + new String(column.value) + ":" + column.timestamp);
|
||||
}
|
||||
System.exit(1);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue