This is an automated email from the ASF dual-hosted git repository. yukon pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/rocketmq-remoting.git
commit 51a189e22125cc6f63ab29a8bac9126b3383b4da Author: yukon <[email protected]> AuthorDate: Thu May 16 14:22:08 2019 +0800 Remove compressor --- .../remoting/api/compression/Compressor.java | 28 ------ .../api/compression/CompressorFactory.java | 28 ------ .../rocketmq/remoting/config/RemotingConfig.java | 10 --- .../impl/compression/CompressorFactoryImpl.java | 68 -------------- .../remoting/impl/compression/GZipCompressor.java | 100 --------------------- 5 files changed, 234 deletions(-) diff --git a/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/compression/Compressor.java b/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/compression/Compressor.java deleted file mode 100644 index d5c378e..0000000 --- a/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/compression/Compressor.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * 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.rocketmq.remoting.api.compression; - -public interface Compressor { - String name(); - - byte type(); - - byte[] compress(final byte[] content) throws Exception; - - byte[] deCompress(final byte[] content) throws Exception; -} diff --git a/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/compression/CompressorFactory.java b/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/compression/CompressorFactory.java deleted file mode 100644 index 4afd599..0000000 --- a/remoting-core/remoting-api/src/main/java/org/apache/rocketmq/remoting/api/compression/CompressorFactory.java +++ /dev/null @@ -1,28 +0,0 @@ -/* - * 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.rocketmq.remoting.api.compression; - -public interface CompressorFactory { - void register(Compressor compressor); - - byte type(String compressionName); - - Compressor get(byte type); - - void clearAll(); -} diff --git a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/config/RemotingConfig.java b/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/config/RemotingConfig.java index 3f5282c..cc81a11 100644 --- a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/config/RemotingConfig.java +++ b/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/config/RemotingConfig.java @@ -19,7 +19,6 @@ package org.apache.rocketmq.remoting.config; import org.apache.commons.lang3.builder.ToStringBuilder; import org.apache.commons.lang3.builder.ToStringStyle; -import org.apache.rocketmq.remoting.impl.compression.GZipCompressor; public class RemotingConfig extends TcpSocketConfig { private int connectionMaxRetries = 3; @@ -36,7 +35,6 @@ public class RemotingConfig extends TcpSocketConfig { private int threadTaskLowWaterMark = 30000; private int threadTaskHighWaterMark = 50000; private int connectionRetryBackoffMillis = 3000; - private String compressorName = GZipCompressor.COMPRESSOR_NAME; private int serviceThreadBlockQueueSize = 50000; private boolean clientNativeEpollEnable = false; private int clientWorkerThreads = 16 + Runtime.getRuntime().availableProcessors() * 2; @@ -145,14 +143,6 @@ public class RemotingConfig extends TcpSocketConfig { this.connectionRetryBackoffMillis = connectionRetryBackoffMillis; } - public String getCompressorName() { - return compressorName; - } - - public void setCompressorName(final String compressorName) { - this.compressorName = compressorName; - } - public int getServiceThreadBlockQueueSize() { return serviceThreadBlockQueueSize; } diff --git a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/compression/CompressorFactoryImpl.java b/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/compression/CompressorFactoryImpl.java deleted file mode 100644 index 40576ad..0000000 --- a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/compression/CompressorFactoryImpl.java +++ /dev/null @@ -1,68 +0,0 @@ -/* - * 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.rocketmq.remoting.impl.compression; - -import org.apache.rocketmq.remoting.api.compression.Compressor; -import org.apache.rocketmq.remoting.api.compression.CompressorFactory; - -public class CompressorFactoryImpl implements CompressorFactory { - private static final int MAX_COUNT = 0x0FF; - private final Compressor[] tables = new Compressor[MAX_COUNT]; - - public CompressorFactoryImpl() { - this.register(new GZipCompressor()); - } - - @Override - public void register(Compressor compressor) { - if (tables[compressor.type() & MAX_COUNT] != null) { - throw new RuntimeException("compressor header's sign is overlapped"); - } - tables[compressor.type() & MAX_COUNT] = compressor; - } - - @Override - public byte type(String compressionName) { - for (Compressor table : this.tables) { - if (table != null) { - if (table.name().equalsIgnoreCase(compressionName)) { - return table.type(); - } - } - } - - throw new IllegalArgumentException(String.format("the compressor: %s not exist", compressionName)); - - } - - @Override - public Compressor get(byte type) { - return tables[type & MAX_COUNT]; - } - - @Override - public void clearAll() { - for (int i = 0; i < this.tables.length; i++) { - this.tables[i] = null; - } - } - - public Compressor[] getTables() { - return tables; - } -} diff --git a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/compression/GZipCompressor.java b/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/compression/GZipCompressor.java deleted file mode 100644 index 53dd4bf..0000000 --- a/remoting-core/remoting-impl/src/main/java/org/apache/rocketmq/remoting/impl/compression/GZipCompressor.java +++ /dev/null @@ -1,100 +0,0 @@ -/* - * 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.rocketmq.remoting.impl.compression; - -import java.io.ByteArrayInputStream; -import java.io.ByteArrayOutputStream; -import java.io.InputStream; -import java.io.OutputStream; -import java.util.zip.GZIPInputStream; -import java.util.zip.GZIPOutputStream; -import org.apache.rocketmq.remoting.api.compression.Compressor; - -public class GZipCompressor implements Compressor { - public static final int BUFFER = 1024; - public static final String COMPRESSOR_NAME = GZipCompressor.class.getSimpleName(); - public static final byte COMPRESSOR_TYPE = 'G'; - - @Override - public String name() { - return COMPRESSOR_NAME; - } - - @Override - public byte type() { - return COMPRESSOR_TYPE; - } - - @Override - public byte[] compress(byte[] content) throws Exception { - if (content == null) - return new byte[0]; - - ByteArrayInputStream bais = new ByteArrayInputStream(content); - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - - compress(bais, baos); - byte[] output = baos.toByteArray(); - baos.flush(); - baos.close(); - bais.close(); - return output; - - } - - private void compress(InputStream is, OutputStream os) throws Exception { - GZIPOutputStream gos = new GZIPOutputStream(os); - - int count; - byte data[] = new byte[BUFFER]; - while ((count = is.read(data, 0, BUFFER)) != -1) { - gos.write(data, 0, count); - } - gos.finish(); - gos.flush(); - gos.close(); - } - - @Override - public byte[] deCompress(byte[] content) throws Exception { - if (content == null) - return new byte[0]; - - ByteArrayInputStream bais = new ByteArrayInputStream(content); - ByteArrayOutputStream baos = new ByteArrayOutputStream(); - - decompress(bais, baos); - content = baos.toByteArray(); - baos.flush(); - baos.close(); - bais.close(); - return content; - } - - private void decompress(InputStream is, OutputStream os) throws Exception { - GZIPInputStream gis = new GZIPInputStream(is); - - int count; - byte data[] = new byte[BUFFER]; - while ((count = gis.read(data, 0, BUFFER)) != -1) { - os.write(data, 0, count); - } - gis.close(); - } - -}
