diskpagestore.java

Wicket一个开发Java Web应用程序框架。它使得开发web应用程序变得容易而轻松。 Wicket利用一个POJO data beans组件使得它可以与任何持久层技术相结合。

	/**
	 * Returns the file name for specified pagemap. If the session folder (folder that contains the
	 * file) does not exist and createSessionFolder is true, the folder will be created.
	 * 
	 * @param sessionId
	 * @param pageMapName
	 * @param createSessionFolder
	 * @return file name for pagemap
	 */
	private String getPageMapFileName(String sessionId, String pageMapName,
		boolean createSessionFolder)
	{
		File sessionFolder = getSessionFolder(sessionId, createSessionFolder);
		return new File(sessionFolder, "pm-" + pageMapName).getAbsolutePath();
	}

	private final int maxSizePerPageMap;

	/**
	 * Return maximum pagemap file size (in bytes).
	 * 
	 * @return
	 */
	protected int getMaxSizePerPageMap()
	{
		return maxSizePerPageMap;
	}

	private final int maxSizePerSession;

	/**
	 * Returns maximum size per session (in bytes). After the session exceeds this size, appropriate
	 * number of last recently used pagemap files will be removed.
	 * 
	 * @return
	 */
	protected int getMaxSizePerSession()
	{
		return maxSizePerSession;
	}

	private final FileChannelPool fileChannelPool;

	private final File fileStoreFolder;

	/**
	 * Returns the "root" file store folder.
	 * 
	 * @return
	 */
	protected File getFileStoreFolder()
	{
		return fileStoreFolder;
	}

	private final String appName;

	/**
	 * Creates a new {@link DiskPageStore} instance.
	 * 
	 * @param fileStoreFolder
	 *            folder in which the session folders containing pagemap files will be stored
	 * @param maxSizePerPagemap
	 *            the maximum size of pagemap file (in bytes)
	 * @param maxSizePerSession
	 *            the maximum size of session (in bytes)
	 * @param fileChannelPoolCapacity
	 *            the maximum number of concurrently opened files (higher number improves
	 *            performance under heavy load).
	 */
	public DiskPageStore(File fileStoreFolder, int maxSizePerPagemap, int maxSizePerSession,
		int fileChannelPoolCapacity)
	{
		maxSizePerPageMap = maxSizePerPagemap;
		this.maxSizePerSession = maxSizePerSession;
		fileChannelPool = new FileChannelPool(fileChannelPoolCapacity);
		this.fileStoreFolder = fileStoreFolder;

		if (maxSizePerSession < maxSizePerPageMap)
		{
			throw new IllegalArgumentException(
				"Provided maximum session size must be bigger than maximum pagemap size");
		}

		this.fileStoreFolder.mkdirs();
		appName = Application.get().getApplicationKey();

		loadIndex();

		initPageSavingThread();
	}

	private void loadIndex()
	{
		File storeFolder = getStoreFolder();
		File index = new File(storeFolder, "DiskPageStoreIndex");
		if (index.exists() && index.length() > 0)
		{
			try
			{
				InputStream stream = new FileInputStream(index);
				ObjectInputStream ois = new ObjectInputStream(stream);
				Map map = (Map)ois.readObject();
				sessionIdToEntryMap = new ConcurrentHashMap(map);
				for (Iterator entries = sessionIdToEntryMap.entrySet().iterator(); entries.hasNext();)
				{
					// initialize the diskPageStore reference
					Entry entry = (Entry)entries.next();
					SessionEntry sessionEntry = (SessionEntry)entry.getValue();
					sessionEntry.diskPageStore = this;
				}
				stream.close();
			}
			catch (Exception e)
			{
				log.error("Couldn't load DiskPageStore index from file " + index + ".", e);
			}
		}
		index.delete();
	}

	private void saveIndex()
	{
		File storeFolder = getStoreFolder();
		if (storeFolder.exists())
		{
			File index = new File(storeFolder, "DiskPageStoreIndex");
			index.delete();
			try
			{
				OutputStream stream = new FileOutputStream(index);
				ObjectOutputStream oos = new ObjectOutputStream(stream);
				oos.writeObject(sessionIdToEntryMap);
				stream.close();
			}
			catch (Exception e)
			{
				log.error("Couldn't write DiskPageStore index to file " + index + ".", e);
			}
		}
	}

	private static File getDefaultFileStoreFolder()
	{
		final File dir = (File)((WebApplication)Application.get()).getServletContext()
			.getAttribute("javax.servlet.context.tempdir");
		if (dir != null)
		{
			return dir;
		}
		else
		{
			try
			{
				return File.createTempFile("file-prefix", null).getParentFile();
			}
			catch (IOException e)
			{
				throw new WicketRuntimeException(e);
			}
		}
	}

	/**
	 * Creates a new {@link DiskPageStore} instance.
	 * 
	 * @param maxSizePerPagemap
	 *            the maximum size of pagemap file (in bytes)
	 * @param maxSizePerSession
	 *            the maximum size of session (in bytes)
	 * @param fileChannelPoolCapacity
	 *            the maximum number of concurrently opened files (higher number improves
	 *            performance under heavy load).
	 * 
	 */
	public DiskPageStore(int maxSizePerPagemap, int maxSizePerSession, int fileChannelPoolCapacity)
	{
		this(getDefaultFileStoreFolder(), maxSizePerPagemap, maxSizePerSession,
			fileChannelPoolCapacity);
	}

	/**
	 * Creates a new {@link DiskPageStore} instance.
	 */
	public DiskPageStore()
	{
		this((int)Bytes.megabytes(10).bytes(), (int)Bytes.megabytes(100).bytes(), 50);
	}

	/**
	 * @see org.apache.wicket.protocol.http.SecondLevelCacheSessionStore.IPageStore#destroy()
	 */
	public void destroy()
	{
		if (!isSynchronous())
		{
			// make sure that all pages are saved in asynchronous mode
			synchronized (pagesToSaveAll)
			{
				for (Iterator i = pagesToSaveAll.entrySet().iterator(); i.hasNext();)
				{
					Entry entry = (Entry)i.next();
					String sessionId = (String)entry.getKey();
					List pages = (List)entry.getValue();
					synchronized (pages)
					{
						flushPagesToSaveList(sessionId, pages);
					}
				}
			}
		}

		saveIndex();

		fileChannelPool.destroy();
		if (pageSavingThread != null)
		{
			pageSavingThread.stop();
		}
	}

	private Map /* <String, SessionEntry> */sessionIdToEntryMap = new ConcurrentHashMap();

	/**
	 * Returns the SessionEntry for session with given id. If the entry does not yet exist and the
	 * createIfDoesNotExist attribute is set, new SessionEntry will be created.
	 * 
	 * @param sessionId
	 * @param createIfDoesNotExist
	 * @return
	 */
	protected SessionEntry getSessionEntry(String sessionId, boolean createIfDoesNotExist)
	{
		SessionEntry entry = (SessionEntry)sessionIdToEntryMap.get(sessionId);
		if (entry == null && createIfDoesNotExist)
		{
			synchronized (sessionIdToEntryMap)
			{
				entry = (SessionEntry)sessionIdToEntryMap.get(sessionId);
				if (entry == null)
				{
					entry = new SessionEntry(this);
					entry.sessionId = sessionId;
					sessionIdToEntryMap.put(sessionId, entry);
				}
			}
		}
		return entry;
	}

	/**
	 * @see org.apache.wicket.protocol.http.SecondLevelCacheSessionStore.IPageStore#getPage(java.lang.String,
	 *      java.lang.String, int, int, int)
	 */
	public Page getPage(String sessionId, String pagemap, int id, int versionNumber,
		int ajaxVersionNumber)
	{
		SessionEntry entry = getSessionEntry(sessionId, false);
		if (entry != null)
		{
			byte[] data;

			if (isSynchronous())
			{
				data = entry.loadPage(pagemap, id, versionNumber, ajaxVersionNumber);
			}
			else
			{
				// we need to make sure that the there are no pending pages to
				// be saved before loading a page
				List pages = getPagesToSaveList(sessionId);
				synchronized (pages)
				{
					flushPagesToSaveList(sessionId, pages);
					data = entry.loadPage(pagemap, id, versionNumber, ajaxVersionNumber);
				}
			}

			if (data != null)
			{
				return deserializePage(data, versionNumber);
			}
		}

		return null;
	}

	/**
	 * @see org.apache.wicket.protocol.http.SecondLevelCacheSessionStore.IPageStore#pageAccessed(java.lang.String,
	 *      org.apache.wicket.Page)
	 */
	public void pageAccessed(String sessionId, Page page)
	{
	}

	/**
	 * Removes the page (or entire pagemap) from specified session.
	 * 
	 * @param entry
	 * @param pageMap
	 * @param id
	 *            page id to remove or -1 if the whole pagemap should be removed
	 */
	private void removePage(SessionEntry entry, String pageMap, int id)
	{
		if (id != -1)
		{
			entry.removePage(pageMap, id);
		}
		else
		{
			entry.removePageMap(pageMap);
		}
	}

	/**
	 * @see org.apache.wicket.protocol.http.SecondLevelCacheSessionStore.IPageStore#removePage(java.lang.String,
	 *      java.lang.String, int)
	 */
	public void removePage(String sessionId, String pageMap, int id)
	{
		SessionEntry entry = getSessionEntry(sessionId, false);
		if (entry != null)
		{
			if (isSynchronous())
			{
				removePage(entry, pageMap, id);
			}
			else
			{
				// we need to make sure that the there are no pending pages to
				// be saved before removing the page (or pagemap)
				List pages = getPagesToSaveList(sessionId);
				synchronized (pages)
				{
					flushPagesToSaveList(sessionId, pages);
					removePage(entry, pageMap, id);
				}
			}
		}
	}

	/**
	 * Stores the serialized pages. The storing is done either immediately (in synchronous mode) or
	 * it's scheduled to be stored by the worker thread.
	 * 
	 * @param sessionId
	 * @param pages
	 */
	protected void storeSerializedPages(String sessionId, List /* <SerializedPage> */pages)
	{
		SessionEntry entry = getSessionEntry(sessionId, true);

		if (isSynchronous())
		{
			for (Iterator i = pages.iterator(); i.hasNext();)
			{
				SerializedPage serializedPage = (SerializedPage)i.next();
				entry.savePage(serializedPage);
			}
		}
		else
		{
			schedulePagesSave(sessionId, pages);
		}
	}

	/**
	 * Hook for processing serialized pages (e.g. sending those across cluster)
	 * 
	 * @param sessionId
	 * @param pages
	 */
	protected void onPagesSerialized(String sessionId, List /* <SerializedPage */pages)
	{

	}

	/**
	 * @see org.apache.wicket.protocol.http.SecondLevelCacheSessionStore.IPageStore#storePage(java.lang.String,
	 *      org.apache.wicket.Page)
	 */
	public void storePage(String sessionId, Page page)
	{
		List pages = serializePage(page);

		serializedPagesCache.storePage(sessionId, page, pages);

		onPagesSerialized(sessionId, pages);

		storeSerializedPages(sessionId, pages);
	}

	/**
	 * @see org.apache.wicket.protocol.http.SecondLevelCacheSessionStore.IPageStore#unbind(java.lang.String)
	 */
	public void unbind(String sessionId)
	{
		SessionEntry entry = (SessionEntry)sessionIdToEntryMap.get(sessionId);
		if (entry != null)
		{
			if (isSynchronous())
			{
				entry.unbind();
			}
			else
			{
				List pages = getPagesToSaveList(sessionId);
				synchronized (pages)
				{
					flushPagesToSaveList(sessionId, pages);
					entry.unbind();
				}
				pagesToSaveAll.remove(sessionId);
			}
		}
	}

	// map from session id to serialized page list
	// this contains lists for all active sessions
	private final Map /* <String, List<SerializedPage>> */pagesToSaveAll = new ConcurrentHashMap();