1515 */
1616package com .datastax .driver .osgi .impl ;
1717
18- import com .datastax .driver .core .Cluster ;
19- import com .datastax .driver .core .CodecRegistry ;
20- import com .datastax .driver .core .Metadata ;
21- import com .datastax .driver .core .Session ;
18+ import com .datastax .driver .core .*;
2219import com .datastax .driver .core .exceptions .InvalidQueryException ;
20+ import com .datastax .driver .core .policies .PercentileSpeculativeExecutionPolicy ;
2321import com .datastax .driver .extras .codecs .date .SimpleTimestampCodec ;
2422import com .datastax .driver .osgi .api .MailboxService ;
2523import org .osgi .framework .BundleActivator ;
2624import org .osgi .framework .BundleContext ;
25+ import org .slf4j .Logger ;
26+ import org .slf4j .LoggerFactory ;
2727
2828import java .util .Hashtable ;
2929
30+ import static com .datastax .driver .core .ProtocolOptions .Compression .LZ4 ;
3031import static com .datastax .driver .osgi .api .MailboxMessage .TABLE ;
3132
3233public class Activator implements BundleActivator {
3334
35+ private static final Logger LOGGER = LoggerFactory .getLogger (Activator .class );
36+
3437 private Cluster cluster ;
3538
3639 @ Override
3740 public void start (BundleContext context ) throws Exception {
41+
42+ VersionNumber ver = VersionNumber .parse (context .getProperty ("cassandra.version" ));
43+ LOGGER .info ("C* version: {}" , ver );
44+
3845 String contactPointsStr = context .getProperty ("cassandra.contactpoints" );
3946 if (contactPointsStr == null ) {
4047 contactPointsStr = "127.0.0.1" ;
4148 }
49+ LOGGER .info ("Contact points: {}" , contactPointsStr );
4250 String [] contactPoints = contactPointsStr .split ("," );
4351
4452 String keyspace = context .getProperty ("cassandra.keyspace" );
4553 if (keyspace == null ) {
4654 keyspace = "mailbox" ;
4755 }
56+ LOGGER .info ("Keyspace: {}" , keyspace );
4857 keyspace = Metadata .quote (keyspace );
4958
50- cluster = Cluster .builder ()
59+ Cluster . Builder builder = Cluster .builder ()
5160 .addContactPoints (contactPoints )
52- .withCodecRegistry (new CodecRegistry ().register (SimpleTimestampCodec .instance ))
53- .build ();
61+ .withCodecRegistry (new CodecRegistry ().register (SimpleTimestampCodec .instance ));
62+
63+ String compression = context .getProperty ("cassandra.compression" );
64+ if (compression != null ) {
65+ if (ver .getMajor () < 2 && compression .equals (LZ4 .name ())) {
66+ LOGGER .warn ("Requested LZ4 compression but C* version is not compatible, disabling" );
67+ } else {
68+ LOGGER .info ("Compression: {}" , compression );
69+ builder .withCompression (ProtocolOptions .Compression .valueOf (compression ));
70+ }
71+ } else {
72+ LOGGER .info ("Compression: NONE" );
73+ }
74+
75+ String usePercentileSpeculativeExecutionPolicy = context .getProperty ("cassandra.usePercentileSpeculativeExecutionPolicy" );
76+ if ("true" .equals (usePercentileSpeculativeExecutionPolicy )) {
77+ PerHostPercentileTracker perHostPercentileTracker = PerHostPercentileTracker .builderWithHighestTrackableLatencyMillis (15000 ).build ();
78+ builder .withSpeculativeExecutionPolicy (new PercentileSpeculativeExecutionPolicy (perHostPercentileTracker , 99 , 1 ));
79+ LOGGER .info ("Use PercentileSpeculativeExecutionPolicy: YES" );
80+ } else {
81+ LOGGER .info ("Use PercentileSpeculativeExecutionPolicy: NO" );
82+ }
83+
84+ cluster = builder .build ();
5485
5586 Session session ;
5687 try {
@@ -73,6 +104,7 @@ public void start(BundleContext context) throws Exception {
73104 mailbox .init ();
74105
75106 context .registerService (MailboxService .class .getName (), mailbox , new Hashtable <String , String >());
107+ LOGGER .info ("Mailbox Service successfully initialized" );
76108 }
77109
78110 @ Override
0 commit comments