import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public Ticket getTicket(final String ticketId) {
    try {
        final SerializableDocument document = couchbase.bucket().get(ticketId, SerializableDocument.class);
        if (document != null) {
            final Ticket t = (Ticket) document.content();
            logger.debug("Got ticket {} from registry.", t);
            return getProxiedTicketInstance(t);
        }
        logger.debug("Ticket {} not found in registry.", ticketId);
        return null;
    } catch (final Exception e) {
        logger.error("Failed fetching {}: {}", ticketId, e);
        return null;
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public Ticket getTicket(final String ticketId) {
    try {
        LOGGER.debug("Locating ticket id [{}]", ticketId);
        final String encTicketId = encodeTicketId(ticketId);
        if (encTicketId == null) {
            LOGGER.debug("Ticket id [{}] could not be found", ticketId);
            return null;
        }

        final SerializableDocument document = this.couchbase.bucket().get(encTicketId, SerializableDocument.class);
        if (document != null) {
            final Ticket t = (Ticket) document.content();
            LOGGER.debug("Got ticket [{}] from the registry.", t);
            return t;
        }
        LOGGER.debug("Ticket [{}] not found in the registry.", encTicketId);
        return null;
    } catch (final Exception e) {
        LOGGER.error("Failed fetching [{}]: [{}]", ticketId, e);
        return null;
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Test
public void testGetAndPut() throws Exception {
    String old = dedicatedCache.getAndPut("testGetAndPut", "getAndPutTest");
    SerializableDocument doc = dedicatedCache.bucket.get(cbKey("testGetAndPut"), SerializableDocument.class);

    assertNotNull(doc);
    assertNotNull(doc.content());
    assertEquals("getAndPutTest", doc.content());
    assertNull(old);

    old = dedicatedCache.getAndPut("testGetAndPut", "getAndPutTest2");
    doc = dedicatedCache.bucket.get(cbKey("testGetAndPut"), SerializableDocument.class);

    assertEquals("getAndPutTest", old);
    assertNotNull(doc);
    assertEquals("getAndPutTest2", doc.content());

}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public void removeAll() {
    checkOpen();
    long start = configuration.isStatisticsEnabled() ? System.nanoTime() : 0;

    final AtomicLong removedCount = new AtomicLong(0L);
    internalClear(new Action1<SerializableDocument>() {
        @Override
        public void call(SerializableDocument serializableDocument) {
            K key = fromInternalKey(serializableDocument.id());
            V value = (V) serializableDocument.content();
            eventManager.queueEvent(new CouchbaseCacheEntryEvent<K, V>(EventType.REMOVED, key, value,
                    CouchbaseCache.this));
        }
    });
    eventManager.dispatch();

    if (configuration.isStatisticsEnabled() && removedCount.get() > 0L) {
        statisticsMxBean.increaseCacheRemovals(removedCount.get());
        //approximate remove time as an average
        statisticsMxBean.addRemoveTimeNano((System.nanoTime() - start) / removedCount.get());
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
/**
 * Depending on the operation, produces a SerializableDocument with correct TTL and CAS.
 *
 * @param key the key for the document
 * @param value the value to store
 * @param op the operation being performed
 * @param cas the cas of the document (or 0 if none needed)
 * @return the {@link SerializableDocument} to be persisted, or null if the {@link ExpiryPolicy}
 *  indicates a TTL already expired
 * @throws IllegalArgumentException when the {@link ExpiryPolicy} produces a TTL > 30 days
 */
private SerializableDocument createDocument(K key, V value, Operation op, long cas) {
    String cbKey = toInternalKey(key);
    Serializable cbValue = toInternalValue(value);
    int ttlOrCode = getDurationCode(op);
    switch (ttlOrCode) {
        case TTL_DONT_CHANGE:
            return SerializableDocument.create(cbKey, cbValue, cas);
        case TTL_NONE:
            return SerializableDocument.create(cbKey, cbValue, cas);
        case TTL_EXPIRED:
            return null;
        default:
            if (ttlOrCode < 0) {
                throw new IllegalArgumentException("Unknown ttl code " + ttlOrCode);
            } else {
                return SerializableDocument.create(cbKey, ttlOrCode, cbValue, cas);
            }
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
protected void updateTicket(final Ticket ticket) {
    logger.debug("Updating ticket {}", ticket);
    try {
        final SerializableDocument document =
                SerializableDocument.create(ticket.getId(), getTimeout(ticket), ticket);
        couchbase.bucket().upsert(document);
    } catch (final Exception e) {
        logger.error("Failed updating {}: {}", ticket, e);
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public void addTicket(final Ticket ticket) {
    logger.debug("Adding ticket {}", ticket);
    try {
        final SerializableDocument document =
                SerializableDocument.create(ticket.getId(), getTimeout(ticket), ticket);
        couchbase.bucket().upsert(document);
    } catch (final Exception e) {
        logger.error("Failed adding {}: {}", ticket, e);
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public Ticket updateTicket(final Ticket ticket) {
    LOGGER.debug("Updating ticket [{}]", ticket);
    try {
        final SerializableDocument document = SerializableDocument.create(ticket.getId(), getTimeToLive(ticket), ticket);

        LOGGER.debug("Upserting document [{}] into couchbase bucket [{}]", document.id(), this.couchbase.bucket().name());
        this.couchbase.bucket().upsert(document);
    } catch (final Exception e) {
        LOGGER.error("Failed updating [{}]: [{}]", ticket, e);
    }
    return ticket;
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public void addTicket(final Ticket ticketToAdd) {
    LOGGER.debug("Adding ticket [{}]", ticketToAdd);
    try {
        final Ticket ticket = encodeTicket(ticketToAdd);
        final SerializableDocument document = SerializableDocument.create(ticket.getId(), getTimeToLive(ticketToAdd), ticket);
        LOGGER.debug("Created document for ticket [{}]. Upserting into bucket [{}]", ticketToAdd, this.couchbase.bucket().name());
        this.couchbase.bucket().upsert(document);
    } catch (final Exception e) {
        LOGGER.error("Failed adding [{}]: [{}]", ticketToAdd, e);
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
private SerializableDocument assertDocument(String expected, String key) {
    SerializableDocument doc = dedicatedCache.bucket.get(cbKey(key), SerializableDocument.class);

    assertNotNull(doc);
    assertNotNull(doc.content());
    assertEquals(expected, doc.content());

    return doc;
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Test
public void testRemove() throws Exception {
    boolean result = dedicatedCache.remove("remove");

    assertTrue(result);
    assertNull(dedicatedCache.bucket.get(cbKey("remove"), SerializableDocument.class));
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Test
public void testRemoveIfOldValueMatches() throws Exception {
    boolean result = dedicatedCache.remove("oldremove", "not");

    assertFalse(result);
    assertDocument(TEST_VALUE, "oldremove");

    result = dedicatedCache.remove("oldremove", TEST_VALUE);

    assertTrue(result);
    assertNull(dedicatedCache.bucket.get(cbKey("oldremove"), SerializableDocument.class));
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Test
public void testRemoveAllFromSet() throws Exception {
    dedicatedCache.bucket.insert(SerializableDocument.create(cbKey("rk1"), TEST_VALUE));
    dedicatedCache.bucket.insert(SerializableDocument.create(cbKey("rk2"), TEST_VALUE));
    Set<String> toRemove = new HashSet<String>(Arrays.asList("rk1", "rk2", "rk3"));

    dedicatedCache.removeAll(toRemove);

    assertNull(dedicatedCache.bucket.get("rk1"));
    assertNull(dedicatedCache.bucket.get("rk2"));
    assertNull(dedicatedCache.bucket.get("rk3"));
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public void put(K key, V value) {
    //TODO check expiry
    checkOpen();
    checkTypes(key, value);
    long start = configuration.isStatisticsEnabled() ? System.nanoTime() : 0;

    try {
        SerializableDocument oldDocument = bucket.get(toInternalKey(key), SerializableDocument.class);
        SerializableDocument doc = createDocument(key, value, Operation.CREATION);
        //Only do something if doc is not null (otherwise it means expiry was already set)
        if (doc != null) {
            bucket.upsert(doc);
            if (oldDocument != null) {
                eventManager.queueAndDispatch(EventType.UPDATED, key, value, (V) oldDocument.content(), this);
            } else {
                eventManager.queueAndDispatch(EventType.CREATED, key, value, this);
            }
            if (configuration.isStatisticsEnabled()) {
                statisticsMxBean.increaseCachePuts(1L);
                statisticsMxBean.addPutTimeNano(System.nanoTime() - start);
            }
        }
    } catch (Exception e) {
        throw new CacheException("Error during put of " + key, e);
    }

}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public V getAndPut(K key, V value) {
    //TODO expiry
    checkOpen();
    checkTypes(key, value);

    long start = configuration.isStatisticsEnabled() ? System.nanoTime() : 0;

    String internalKey = toInternalKey(key);
    SerializableDocument oldValue = bucket.get(internalKey, SerializableDocument.class);

    put(key, value);
    if (configuration.isStatisticsEnabled()) {
        statisticsMxBean.increaseCachePuts(1L);
        if (oldValue == null) {
            statisticsMxBean.increaseCacheMisses(1L);
        } else {
            statisticsMxBean.increaseCacheHits(1L);
        }
        long time = System.nanoTime() - start;
        statisticsMxBean.addGetTimeNano(time);
        statisticsMxBean.addPutTimeNano(time);
    }

    if (oldValue == null) {
        eventManager.queueAndDispatch(EventType.CREATED, key, value, this);
        return null;
    } else {
        V old = (V) oldValue.content();
        eventManager.queueAndDispatch(EventType.UPDATED, key, value, old, this);
        return old;
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
private void internalClear(Action1<? super SerializableDocument> action) {
    getAllKeys()
            .flatMap(new Func1<String, Observable<SerializableDocument>>() {
                @Override
                public Observable<SerializableDocument> call(String id) {
                    return bucket.async().remove(id, SerializableDocument.class);
                }
            })
            .toBlocking()
            .forEach(action);
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
private Notification<? extends SerializableDocument> take() {
    try {
        return notifications.take();
    } catch (InterruptedException e) {
        throw Exceptions.propagate(e);
    }
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@BeforeClass
public static void init() {
    //here is where you'd get the environment
    cachingProvider = new CouchbaseCachingProvider();
    //here is where we get the cluster
    cacheManager = (CouchbaseCacheManager) cachingProvider.getCacheManager();

    ClusterManager manager = cacheManager.cluster.clusterManager(CLUSTER_LOGIN, CLUSTER_PWD);
    boolean hasBucket = manager.hasBucket(CACHE_NAME);
    if (!hasBucket) {
        manager.insertBucket(DefaultBucketSettings.builder()
                                                  .name(CACHE_NAME)
                                                  .enableFlush(true)
                                                  .quota(100)
                                                  .build());
    }

    //here we create a dedicated cache for most of the tests
    CouchbaseConfiguration<String, String> cbConfig = new CouchbaseConfiguration.Builder<String, String>(
            CACHE_NAME, KeyConverter.STRING_KEY_CONVERTER)
            .useDedicatedBucket(CACHE_NAME, "")
            .build();
    dedicatedCache = (CouchbaseCache<String, String>) cacheManager.createCache(CACHE_NAME, cbConfig);
    keyConverter = dedicatedCache.keyConverter();

    //prepare data
    if (hasBucket) {
        dedicatedCache.bucket.bucketManager().flush();
    }
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString(TEST_KEY), TEST_VALUE));
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString(TEST_KEY) + "2", TEST_VALUE + "2"));
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString("remove"), TEST_VALUE));
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString("oldremove"), TEST_VALUE));
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString("getremove"), TEST_VALUE));
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString("itremove"), TEST_VALUE));
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString("replace"), TEST_VALUE));
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString("oldreplace"), TEST_VALUE));
    dedicatedCache.bucket.insert(SerializableDocument.create(keyConverter.asString("getreplace"), TEST_VALUE));

    //TODO prepare view
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
private void internalReplace(K key, V value, V oldValue, String cbKey, Serializable internalValue, long cas) {
    SerializableDocument newDoc = SerializableDocument.create(cbKey, internalValue, cas);
    bucket.replace(newDoc);
    eventManager.queueAndDispatch(EventType.UPDATED, key, value, oldValue, this);
}
 

import com.couchbase.client.java.document.SerializableDocument; //导入依赖的package包/类
@Override
public Tuple2<Long, SerializableDocument> call(Long aLong, SerializableDocument serializableDocument) {
    return Tuple.create(aLong, serializableDocument);
}